feat(cli): move shared destination lifecycle to Rust

Move the warning, destination open/attach, shared file iteration, and close ordering for multi-file compression and decompression into Rust orchestration. The C side retains the private preferences/resources, filesystem callbacks, write-pool operations, diagnostics, and exception behavior behind opaque callbacks. Decompression test mode continues to skip destination I/O. Add explicit C/Rust projection layout assertions and focused lifecycle tests covering callback order, open failure, aggregate file errors, and test mode. Test Plan: - git diff --check - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo check --manifest-path rust/Cargo.toml --tests - ulimit -v 41943040; make -j1 - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings - capped two-input shared-destination compression/decompression smoke test
2026-07-21 20:09:20 +02:00
parent 7cc25be2a8
commit 073f198c53
2 changed files with 552 additions and 44 deletions
@@ -602,6 +602,14 @@ pub type FIO_rust_compress_multiple_file_fn =
    unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char) -> c_int;
 pub type FIO_rust_compress_multiple_separate_file_fn =
    unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int;
+pub type FIO_rust_multiple_destination_warning_fn =
+    unsafe extern "C" fn(*mut c_void) -> c_int;
+pub type FIO_rust_multiple_destination_open_fn =
+    unsafe extern "C" fn(*mut c_void) -> *mut c_void;
+pub type FIO_rust_multiple_destination_attach_fn =
+    unsafe extern "C" fn(*mut c_void, *mut c_void);
+pub type FIO_rust_multiple_destination_close_fn =
+    unsafe extern "C" fn(*mut c_void) -> c_int;

 /// Rust owns only the shared-destination file iteration and aggregate error
 /// handling.  C retains the private preferences/resources and supplies one
@@ -636,6 +644,52 @@ const _: () = {
    assert!(size_of::<FIO_rust_compress_multiple_projection_t>() == 5 * size_of::<usize>());
 };

+/// Rust owns the shared-destination lifecycle around the file iterator. C
+/// retains warning, filesystem, and write-pool operations behind callbacks.
+#[repr(C)]
+pub struct FIO_rust_compress_multiple_destination_projection_t {
+    pub files: *const FIO_rust_compress_multiple_projection_t,
+    pub warning: Option<FIO_rust_multiple_destination_warning_fn>,
+    pub open_destination: Option<FIO_rust_multiple_destination_open_fn>,
+    pub attach_destination: Option<FIO_rust_multiple_destination_attach_fn>,
+    pub close_destination: Option<FIO_rust_multiple_destination_close_fn>,
+}
+const _: () = {
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_destination_projection_t, files) == 0
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_destination_projection_t, warning)
+            == size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_compress_multiple_destination_projection_t,
+            open_destination
+        ) == 2 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_compress_multiple_destination_projection_t,
+            attach_destination
+        ) == 3 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_compress_multiple_destination_projection_t,
+            close_destination
+        ) == 4 * size_of::<usize>()
+    );
+    assert!(size_of::<FIO_rust_multiple_destination_warning_fn>() == size_of::<usize>());
+    assert!(size_of::<FIO_rust_multiple_destination_open_fn>() == size_of::<usize>());
+    assert!(size_of::<FIO_rust_multiple_destination_attach_fn>() == size_of::<usize>());
+    assert!(size_of::<FIO_rust_multiple_destination_close_fn>() == size_of::<usize>());
+    assert!(
+        size_of::<FIO_rust_compress_multiple_destination_projection_t>()
+            == 5 * size_of::<usize>()
+    );
+};
+
 /// Rust owns separate-destination mode selection, file iteration, and
 /// aggregate error handling. C retains destination-name construction, private
 /// preferences/resources, diagnostics, and compression dispatch in the two
@@ -724,6 +778,54 @@ const _: () = {
    assert!(size_of::<FIO_rust_decompress_multiple_projection_t>() == 5 * size_of::<usize>());
 };

+/// Rust owns the shared-destination lifecycle around the decompression file
+/// iterator. Test mode skips the destination callbacks just as the original
+/// C path did.
+#[repr(C)]
+pub struct FIO_rust_decompress_multiple_destination_projection_t {
+    pub files: *const FIO_rust_decompress_multiple_projection_t,
+    pub destination_enabled: c_int,
+    pub warning: Option<FIO_rust_multiple_destination_warning_fn>,
+    pub open_destination: Option<FIO_rust_multiple_destination_open_fn>,
+    pub attach_destination: Option<FIO_rust_multiple_destination_attach_fn>,
+    pub close_destination: Option<FIO_rust_multiple_destination_close_fn>,
+}
+const _: () = {
+    assert!(
+        std::mem::offset_of!(FIO_rust_decompress_multiple_destination_projection_t, files) == 0
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_decompress_multiple_destination_projection_t,
+            destination_enabled
+        ) == size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_decompress_multiple_destination_projection_t, warning)
+            == 2 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_decompress_multiple_destination_projection_t,
+            open_destination
+        ) == 3 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_decompress_multiple_destination_projection_t,
+            attach_destination
+        ) == 4 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(
+            FIO_rust_decompress_multiple_destination_projection_t,
+            close_destination
+        ) == 5 * size_of::<usize>()
+    );
+    assert!(size_of::<c_int>() <= size_of::<usize>());
+    assert!(size_of::<FIO_rust_decompress_multiple_destination_projection_t>() == 6 * size_of::<usize>());
+};
+
 /// Rust owns only the separate-destination file iteration and aggregate error
 /// handling. C retains destination-name construction, source opening, format
 /// dispatch, diagnostics, and source removal through the per-source callback.
@@ -3372,6 +3474,49 @@ pub unsafe extern "C" fn FIO_rust_compressMultipleFilenames(
    error
 }

+/// Open, attach, iterate, and close one destination shared by all input
+/// files. C retains the private destination and write-pool operations.
+#[no_mangle]
+pub unsafe extern "C" fn FIO_rust_compressMultipleFilenamesWithDestination(
+    projection: *const FIO_rust_compress_multiple_destination_projection_t,
+) -> c_int {
+    assert!(!projection.is_null());
+    let projection = unsafe { &*projection };
+    assert!(!projection.files.is_null());
+    let files = unsafe { &*projection.files };
+    assert!(!files.f_ctx.is_null());
+    assert!(!files.input_file_names.is_null());
+    assert!(!files.output_file_name.is_null());
+    assert!(!files.opaque.is_null());
+    assert!(files.compress_file.is_some());
+
+    let warning = projection
+        .warning
+        .expect("shared-destination warning callback is required");
+    if unsafe { warning(files.opaque) } != 0 {
+        return 1;
+    }
+
+    let open_destination = projection
+        .open_destination
+        .expect("shared-destination open callback is required");
+    let destination = unsafe { open_destination(files.opaque) };
+    if destination.is_null() {
+        return 1;
+    }
+
+    let attach_destination = projection
+        .attach_destination
+        .expect("shared-destination attach callback is required");
+    unsafe { attach_destination(files.opaque, destination) };
+
+    let result = unsafe { FIO_rust_compressMultipleFilenames(projection.files) };
+    let close_destination = projection
+        .close_destination
+        .expect("shared-destination close callback is required");
+    result | unsafe { close_destination(files.opaque) }
+}
+
 /// Select the destination-mode callback once, then iterate files that each
 /// receive a separate destination. C retains destination-name construction,
 /// resource state, diagnostics, and compression dispatch; Rust preserves the
@@ -3449,6 +3594,53 @@ pub unsafe extern "C" fn FIO_rust_decompressMultipleFilenames(
    error
 }

+/// Run the shared-destination decompression lifecycle. Test mode still runs
+/// the warning and file iterator, but intentionally skips destination I/O.
+#[no_mangle]
+pub unsafe extern "C" fn FIO_rust_decompressMultipleFilenamesWithDestination(
+    projection: *const FIO_rust_decompress_multiple_destination_projection_t,
+) -> c_int {
+    assert!(!projection.is_null());
+    let projection = unsafe { &*projection };
+    assert!(!projection.files.is_null());
+    let files = unsafe { &*projection.files };
+    assert!(!files.f_ctx.is_null());
+    assert!(!files.src_names_table.is_null());
+    assert!(!files.out_file_name.is_null());
+    assert!(!files.opaque.is_null());
+    assert!(files.decompress_file.is_some());
+
+    let warning = projection
+        .warning
+        .expect("shared-destination warning callback is required");
+    if unsafe { warning(files.opaque) } != 0 {
+        return 1;
+    }
+
+    if projection.destination_enabled == 0 {
+        return unsafe { FIO_rust_decompressMultipleFilenames(projection.files) };
+    }
+
+    let open_destination = projection
+        .open_destination
+        .expect("shared-destination open callback is required");
+    let destination = unsafe { open_destination(files.opaque) };
+    if destination.is_null() {
+        return 1;
+    }
+
+    let attach_destination = projection
+        .attach_destination
+        .expect("shared-destination attach callback is required");
+    unsafe { attach_destination(files.opaque, destination) };
+
+    let result = unsafe { FIO_rust_decompressMultipleFilenames(projection.files) };
+    let close_destination = projection
+        .close_destination
+        .expect("shared-destination close callback is required");
+    result | unsafe { close_destination(files.opaque) }
+}
+
 /// Iterate files that each receive a separate destination. The C callback
 /// retains destination-name construction, private I/O, format dispatch, and
 /// diagnostics; Rust owns per-file ordering and removal policy.
@@ -7240,6 +7432,191 @@ mod tests {
        status
    }

+    #[derive(Default)]
+    struct MultipleDestinationSessionState {
+        events: Vec<&'static str>,
+        sources: Vec<*const c_char>,
+        destinations: Vec<*const c_char>,
+        statuses: Vec<c_int>,
+        allow_open: bool,
+        warning_status: c_int,
+        close_status: c_int,
+    }
+
+    unsafe extern "C" fn record_multiple_destination_warning(opaque: *mut c_void) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<MultipleDestinationSessionState>() };
+        state.events.push("warning");
+        state.warning_status
+    }
+
+    unsafe extern "C" fn record_multiple_destination_open(opaque: *mut c_void) -> *mut c_void {
+        let state = unsafe { &mut *opaque.cast::<MultipleDestinationSessionState>() };
+        state.events.push("open");
+        if state.allow_open {
+            ptr::dangling_mut::<c_void>()
+        } else {
+            ptr::null_mut()
+        }
+    }
+
+    unsafe extern "C" fn record_multiple_destination_attach(
+        opaque: *mut c_void,
+        destination: *mut c_void,
+    ) {
+        assert!(!destination.is_null());
+        let state = unsafe { &mut *opaque.cast::<MultipleDestinationSessionState>() };
+        state.events.push("attach");
+    }
+
+    unsafe extern "C" fn record_multiple_destination_close(opaque: *mut c_void) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<MultipleDestinationSessionState>() };
+        state.events.push("close");
+        state.close_status
+    }
+
+    unsafe extern "C" fn record_multiple_destination_file(
+        opaque: *mut c_void,
+        destination: *const c_char,
+        source: *const c_char,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<MultipleDestinationSessionState>() };
+        let status = state
+            .statuses
+            .get(state.sources.len())
+            .copied()
+            .unwrap_or(0);
+        state.events.push("file");
+        state.sources.push(source);
+        state.destinations.push(destination);
+        status
+    }
+
+    #[test]
+    fn compression_multiple_destination_session_owns_lifecycle() {
+        let source_names = [c"one".as_ptr(), c"two".as_ptr()];
+        let destination = c"archive.zst";
+        let mut context = FIO_rust_compression_context_t {
+            nbFilesTotal: 2,
+            hasStdinInput: 0,
+            hasStdoutOutput: 0,
+            currFileIdx: 0,
+            nbFilesProcessed: 0,
+            totalBytesInput: 0,
+            totalBytesOutput: 0,
+        };
+        let mut state = MultipleDestinationSessionState {
+            allow_open: true,
+            statuses: vec![0, 4],
+            ..MultipleDestinationSessionState::default()
+        };
+        let files = 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 MultipleDestinationSessionState).cast(),
+            compress_file: Some(record_multiple_destination_file),
+        };
+        let projection = FIO_rust_compress_multiple_destination_projection_t {
+            files: &files,
+            warning: Some(record_multiple_destination_warning),
+            open_destination: Some(record_multiple_destination_open),
+            attach_destination: Some(record_multiple_destination_attach),
+            close_destination: Some(record_multiple_destination_close),
+        };
+
+        assert_eq!(
+            unsafe { FIO_rust_compressMultipleFilenamesWithDestination(&projection) },
+            4
+        );
+        assert_eq!(state.events, ["warning", "open", "attach", "file", "file", "close"]);
+        assert_eq!(state.sources, source_names);
+        assert_eq!(state.destinations, vec![destination.as_ptr(); 2]);
+        assert_eq!(context.currFileIdx, 2);
+        assert_eq!(context.nbFilesProcessed, 1);
+    }
+
+    #[test]
+    fn compression_multiple_destination_session_stops_on_open_failure() {
+        let source_names = [c"one".as_ptr()];
+        let destination = c"archive.zst";
+        let mut context = FIO_rust_compression_context_t {
+            nbFilesTotal: 1,
+            hasStdinInput: 0,
+            hasStdoutOutput: 0,
+            currFileIdx: 0,
+            nbFilesProcessed: 3,
+            totalBytesInput: 0,
+            totalBytesOutput: 0,
+        };
+        let mut state = MultipleDestinationSessionState::default();
+        let files = 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 MultipleDestinationSessionState).cast(),
+            compress_file: Some(record_multiple_destination_file),
+        };
+        let projection = FIO_rust_compress_multiple_destination_projection_t {
+            files: &files,
+            warning: Some(record_multiple_destination_warning),
+            open_destination: Some(record_multiple_destination_open),
+            attach_destination: Some(record_multiple_destination_attach),
+            close_destination: Some(record_multiple_destination_close),
+        };
+
+        assert_eq!(
+            unsafe { FIO_rust_compressMultipleFilenamesWithDestination(&projection) },
+            1
+        );
+        assert_eq!(state.events, ["warning", "open"]);
+        assert!(state.sources.is_empty());
+        assert_eq!(context.currFileIdx, 0);
+        assert_eq!(context.nbFilesProcessed, 3);
+    }
+
+    #[test]
+    fn decompression_multiple_destination_session_skips_io_in_test_mode() {
+        let source_names = [c"one".as_ptr(), c"two".as_ptr()];
+        let destination = c"archive";
+        let mut context = FIO_rust_compression_context_t {
+            nbFilesTotal: 2,
+            hasStdinInput: 0,
+            hasStdoutOutput: 0,
+            currFileIdx: 0,
+            nbFilesProcessed: 0,
+            totalBytesInput: 0,
+            totalBytesOutput: 0,
+        };
+        let mut state = MultipleDestinationSessionState {
+            statuses: vec![0, 8],
+            ..MultipleDestinationSessionState::default()
+        };
+        let files = FIO_rust_decompress_multiple_projection_t {
+            f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(),
+            src_names_table: source_names.as_ptr(),
+            out_file_name: destination.as_ptr(),
+            opaque: (&mut state as *mut MultipleDestinationSessionState).cast(),
+            decompress_file: Some(record_multiple_destination_file),
+        };
+        let projection = FIO_rust_decompress_multiple_destination_projection_t {
+            files: &files,
+            destination_enabled: 0,
+            warning: Some(record_multiple_destination_warning),
+            open_destination: Some(record_multiple_destination_open),
+            attach_destination: Some(record_multiple_destination_attach),
+            close_destination: Some(record_multiple_destination_close),
+        };
+
+        assert_eq!(
+            unsafe { FIO_rust_decompressMultipleFilenamesWithDestination(&projection) },
+            8
+        );
+        assert_eq!(state.events, ["warning", "file", "file"]);
+        assert_eq!(state.sources, source_names);
+        assert_eq!(context.currFileIdx, 2);
+        assert_eq!(context.nbFilesProcessed, 1);
+    }
+
    #[test]
    fn compression_multiple_separate_destinations_selects_mirror_callback_only() {
        let source_names = [c"one".as_ptr(), c"two".as_ptr()];