refactor(fileio): move compression destination policy to Rust

Move the single-file compression destination lifecycle into the Rust file-I/O policy layer while keeping C responsible for private resources, file handles, diagnostics, metadata operations, and codec dispatch callbacks. Preserve the original shared-destination fast path, temporary permissions, handler timing, metadata-before-close ordering, close-error propagation, stdout cleanup guard, and failed-artifact removal. Test Plan: - cargo fmt --manifest-path rust/Cargo.toml --all -- --check - git diff --check - Full capped Rust, native, CLI, and upstream test suites to follow
2026-07-20 03:32:46 +02:00
parent ef50115fcb
commit 4b96b32118
2 changed files with 627 additions and 54 deletions
@@ -365,6 +365,115 @@ const _: () = {
    );
 };

+pub const FIO_RUST_COMPRESS_DST_OPEN_OK: c_int = 0;
+
+pub type FIO_rust_compress_dst_open_fn =
+    unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int, *mut c_int) -> c_int;
+pub type FIO_rust_compress_dst_handler_fn = unsafe extern "C" fn(*mut c_void);
+pub type FIO_rust_compress_dst_file_fn =
+    unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int) -> c_int;
+pub type FIO_rust_compress_dst_stat_fn = unsafe extern "C" fn(*mut c_void, c_int, *const c_char);
+pub type FIO_rust_compress_dst_close_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
+pub type FIO_rust_compress_dst_remove_fn = unsafe extern "C" fn(*mut c_void, *const c_char);
+
+/// Rust owns the per-file compression destination lifecycle.  C retains the
+/// private resource and file handles, diagnostics, metadata operations, and
+/// format dispatch behind opaque callbacks.
+#[repr(C)]
+pub struct FIO_rust_compress_dst_projection_t {
+    pub opaque: *mut c_void,
+    pub dst_file_name: *const c_char,
+    pub src_file_name: *const c_char,
+    pub compression_level: c_int,
+    pub destination_already_open: c_int,
+    pub source_is_stdin: c_int,
+    pub destination_is_stdout: c_int,
+    pub source_is_regular: c_int,
+    pub open_destination: Option<FIO_rust_compress_dst_open_fn>,
+    pub attach_destination: Option<FIO_rust_compress_dst_handler_fn>,
+    pub add_handler: Option<FIO_rust_compress_dst_handler_fn>,
+    pub compress: Option<FIO_rust_compress_dst_file_fn>,
+    pub clear_handler: Option<FIO_rust_compress_dst_handler_fn>,
+    pub set_fd_stat: Option<FIO_rust_compress_dst_stat_fn>,
+    pub close_destination: Option<FIO_rust_compress_dst_close_fn>,
+    pub utime_destination: Option<FIO_rust_compress_dst_handler_fn>,
+    pub remove_destination: Option<FIO_rust_compress_dst_remove_fn>,
+}
+const _: () = {
+    let callback_offset = (3 * size_of::<usize>() + 5 * size_of::<c_int>() + size_of::<usize>()
+        - 1)
+        / size_of::<usize>()
+        * size_of::<usize>();
+    assert!(std::mem::offset_of!(FIO_rust_compress_dst_projection_t, opaque) == 0);
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, dst_file_name)
+            == size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, src_file_name)
+            == 2 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, compression_level)
+            == 3 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, destination_already_open)
+            == 3 * size_of::<usize>() + size_of::<c_int>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, source_is_stdin)
+            == 3 * size_of::<usize>() + 2 * size_of::<c_int>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, destination_is_stdout)
+            == 3 * size_of::<usize>() + 3 * size_of::<c_int>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, source_is_regular)
+            == 3 * size_of::<usize>() + 4 * size_of::<c_int>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, open_destination)
+            == callback_offset
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, attach_destination)
+            == callback_offset + size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, add_handler)
+            == callback_offset + 2 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, compress)
+            == callback_offset + 3 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, clear_handler)
+            == callback_offset + 4 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, set_fd_stat)
+            == callback_offset + 5 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, close_destination)
+            == callback_offset + 6 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, utime_destination)
+            == callback_offset + 7 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_dst_projection_t, remove_destination)
+            == callback_offset + 8 * size_of::<usize>()
+    );
+    assert!(
+        size_of::<FIO_rust_compress_dst_projection_t>() == callback_offset + 9 * size_of::<usize>()
+    );
+};
+
 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 =
@@ -2616,6 +2725,113 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameSrcFile(
    result
 }

+unsafe fn compress_destination(projection: &FIO_rust_compress_dst_projection_t) -> c_int {
+    let compress = projection
+        .compress
+        .expect("compression callback is required");
+    if projection.destination_already_open != 0 {
+        return unsafe {
+            compress(
+                projection.opaque,
+                projection.dst_file_name,
+                projection.src_file_name,
+                projection.compression_level,
+            )
+        };
+    }
+
+    let transfer_stat = projection.source_is_stdin == 0
+        && projection.destination_is_stdout == 0
+        && projection.source_is_regular != 0;
+    let mut destination_fd = -1;
+    let open_destination = projection
+        .open_destination
+        .expect("destination open callback is required");
+    let open_status = unsafe {
+        open_destination(
+            projection.opaque,
+            projection.src_file_name,
+            projection.dst_file_name,
+            c_int::from(transfer_stat),
+            &mut destination_fd,
+        )
+    };
+    if open_status != FIO_RUST_COMPRESS_DST_OPEN_OK {
+        return 1;
+    }
+
+    let attach_destination = projection
+        .attach_destination
+        .expect("destination attachment callback is required");
+    unsafe { attach_destination(projection.opaque) };
+
+    let add_handler = projection
+        .add_handler
+        .expect("destination handler callback is required");
+    unsafe { add_handler(projection.opaque) };
+
+    let mut result = unsafe {
+        compress(
+            projection.opaque,
+            projection.dst_file_name,
+            projection.src_file_name,
+            projection.compression_level,
+        )
+    };
+
+    let clear_handler = projection
+        .clear_handler
+        .expect("handler-clear callback is required");
+    unsafe { clear_handler(projection.opaque) };
+
+    if transfer_stat {
+        let set_fd_stat = projection
+            .set_fd_stat
+            .expect("destination stat callback is required");
+        unsafe { set_fd_stat(projection.opaque, destination_fd, projection.dst_file_name) };
+    }
+
+    let close_destination = projection
+        .close_destination
+        .expect("destination close callback is required");
+    if unsafe { close_destination(projection.opaque) } != 0 {
+        result = 1;
+    }
+
+    if transfer_stat {
+        let utime_destination = projection
+            .utime_destination
+            .expect("destination timestamp callback is required");
+        unsafe { utime_destination(projection.opaque) };
+    }
+
+    if result != 0 && projection.destination_is_stdout == 0 {
+        let remove_destination = projection
+            .remove_destination
+            .expect("destination removal callback is required");
+        unsafe { remove_destination(projection.opaque, projection.dst_file_name) };
+    }
+
+    result
+}
+
+/// Runs the per-file compression destination policy around C-owned resources
+/// and the existing format-dispatch callback.  The callback order mirrors the
+/// original `FIO_compressFilename_dstFile()` lifecycle, including shared
+/// destinations, metadata, close errors, and failed-output cleanup.
+#[no_mangle]
+pub unsafe extern "C" fn FIO_rust_compressFilenameDstFile(
+    projection: *const FIO_rust_compress_dst_projection_t,
+) -> c_int {
+    assert!(!projection.is_null());
+    let projection = unsafe { &*projection };
+    assert!(!projection.opaque.is_null());
+    assert!(!projection.dst_file_name.is_null());
+    assert!(!projection.src_file_name.is_null());
+
+    unsafe { compress_destination(projection) }
+}
+
 unsafe fn decompress_destination(
    projection: &FIO_rust_decompress_file_projection_t,
    source_is_regular: c_int,
@@ -5012,6 +5228,211 @@ mod tests {
        );
    }

+    const COMPRESS_DST_POLICY_OPEN: u8 = 1;
+    const COMPRESS_DST_POLICY_ATTACH: u8 = 2;
+    const COMPRESS_DST_POLICY_ADD_HANDLER: u8 = 3;
+    const COMPRESS_DST_POLICY_COMPRESS: u8 = 4;
+    const COMPRESS_DST_POLICY_CLEAR_HANDLER: u8 = 5;
+    const COMPRESS_DST_POLICY_SET_FD_STAT: u8 = 6;
+    const COMPRESS_DST_POLICY_CLOSE: u8 = 7;
+    const COMPRESS_DST_POLICY_UTIME: u8 = 8;
+    const COMPRESS_DST_POLICY_REMOVE: u8 = 9;
+
+    #[derive(Default)]
+    struct CompressDestinationPolicyState {
+        events: Vec<u8>,
+        transfer_stats: Vec<c_int>,
+        destination_fds: Vec<c_int>,
+        compression_levels: Vec<c_int>,
+        open_status: c_int,
+        close_status: c_int,
+        compression_status: c_int,
+    }
+
+    unsafe extern "C" fn compress_destination_policy_open(
+        opaque: *mut c_void,
+        _src_file_name: *const c_char,
+        _dst_file_name: *const c_char,
+        transfer_stat: c_int,
+        destination_fd: *mut c_int,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_OPEN);
+        state.transfer_stats.push(transfer_stat);
+        unsafe { *destination_fd = 41 };
+        state.open_status
+    }
+
+    unsafe extern "C" fn compress_destination_policy_handler(opaque: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_ATTACH);
+    }
+
+    unsafe extern "C" fn compress_destination_policy_add_handler(opaque: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_ADD_HANDLER);
+    }
+
+    unsafe extern "C" fn compress_destination_policy_clear_handler(opaque: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_CLEAR_HANDLER);
+    }
+
+    unsafe extern "C" fn compress_destination_policy_compress(
+        opaque: *mut c_void,
+        _dst_file_name: *const c_char,
+        _src_file_name: *const c_char,
+        compression_level: c_int,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_COMPRESS);
+        state.compression_levels.push(compression_level);
+        state.compression_status
+    }
+
+    unsafe extern "C" fn compress_destination_policy_set_fd_stat(
+        opaque: *mut c_void,
+        destination_fd: c_int,
+        _dst_file_name: *const c_char,
+    ) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_SET_FD_STAT);
+        state.destination_fds.push(destination_fd);
+    }
+
+    unsafe extern "C" fn compress_destination_policy_close(opaque: *mut c_void) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_CLOSE);
+        state.close_status
+    }
+
+    unsafe extern "C" fn compress_destination_policy_utime(opaque: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_UTIME);
+    }
+
+    unsafe extern "C" fn compress_destination_policy_remove(
+        opaque: *mut c_void,
+        _dst_file_name: *const c_char,
+    ) {
+        let state = unsafe { &mut *opaque.cast::<CompressDestinationPolicyState>() };
+        state.events.push(COMPRESS_DST_POLICY_REMOVE);
+    }
+
+    fn compress_destination_policy_projection(
+        state: &mut CompressDestinationPolicyState,
+        destination_already_open: c_int,
+        source_is_stdin: c_int,
+        destination_is_stdout: c_int,
+        source_is_regular: c_int,
+    ) -> FIO_rust_compress_dst_projection_t {
+        FIO_rust_compress_dst_projection_t {
+            opaque: (state as *mut CompressDestinationPolicyState).cast(),
+            dst_file_name: c"destination".as_ptr(),
+            src_file_name: c"source".as_ptr(),
+            compression_level: 7,
+            destination_already_open,
+            source_is_stdin,
+            destination_is_stdout,
+            source_is_regular,
+            open_destination: Some(compress_destination_policy_open),
+            attach_destination: Some(compress_destination_policy_handler),
+            add_handler: Some(compress_destination_policy_add_handler),
+            compress: Some(compress_destination_policy_compress),
+            clear_handler: Some(compress_destination_policy_clear_handler),
+            set_fd_stat: Some(compress_destination_policy_set_fd_stat),
+            close_destination: Some(compress_destination_policy_close),
+            utime_destination: Some(compress_destination_policy_utime),
+            remove_destination: Some(compress_destination_policy_remove),
+        }
+    }
+
+    #[test]
+    fn compression_destination_policy_preserves_metadata_and_cleanup_order() {
+        let mut state = CompressDestinationPolicyState::default();
+        let projection = compress_destination_policy_projection(&mut state, 0, 0, 0, 1);
+
+        assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 0);
+        assert_eq!(state.transfer_stats, vec![1]);
+        assert_eq!(state.destination_fds, vec![41]);
+        assert_eq!(state.compression_levels, vec![7]);
+        assert_eq!(
+            state.events,
+            vec![
+                COMPRESS_DST_POLICY_OPEN,
+                COMPRESS_DST_POLICY_ATTACH,
+                COMPRESS_DST_POLICY_ADD_HANDLER,
+                COMPRESS_DST_POLICY_COMPRESS,
+                COMPRESS_DST_POLICY_CLEAR_HANDLER,
+                COMPRESS_DST_POLICY_SET_FD_STAT,
+                COMPRESS_DST_POLICY_CLOSE,
+                COMPRESS_DST_POLICY_UTIME,
+            ]
+        );
+    }
+
+    #[test]
+    fn compression_destination_policy_propagates_close_error_and_removes_output() {
+        let mut state = CompressDestinationPolicyState {
+            close_status: 1,
+            ..CompressDestinationPolicyState::default()
+        };
+        let projection = compress_destination_policy_projection(&mut state, 0, 0, 0, 1);
+
+        assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1);
+        assert_eq!(
+            state.events,
+            vec![
+                COMPRESS_DST_POLICY_OPEN,
+                COMPRESS_DST_POLICY_ATTACH,
+                COMPRESS_DST_POLICY_ADD_HANDLER,
+                COMPRESS_DST_POLICY_COMPRESS,
+                COMPRESS_DST_POLICY_CLEAR_HANDLER,
+                COMPRESS_DST_POLICY_SET_FD_STAT,
+                COMPRESS_DST_POLICY_CLOSE,
+                COMPRESS_DST_POLICY_UTIME,
+                COMPRESS_DST_POLICY_REMOVE,
+            ]
+        );
+    }
+
+    #[test]
+    fn compression_destination_policy_skips_lifecycle_for_shared_destination() {
+        let mut state = CompressDestinationPolicyState {
+            compression_status: 1,
+            ..CompressDestinationPolicyState::default()
+        };
+        let projection = compress_destination_policy_projection(&mut state, 1, 0, 0, 1);
+
+        assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1);
+        assert_eq!(state.events, vec![COMPRESS_DST_POLICY_COMPRESS]);
+        assert!(state.transfer_stats.is_empty());
+        assert!(state.destination_fds.is_empty());
+    }
+
+    #[test]
+    fn compression_destination_policy_keeps_stdout_cleanup_guard() {
+        let mut state = CompressDestinationPolicyState {
+            compression_status: 1,
+            ..CompressDestinationPolicyState::default()
+        };
+        let projection = compress_destination_policy_projection(&mut state, 0, 0, 1, 1);
+
+        assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1);
+        assert_eq!(state.transfer_stats, vec![0]);
+        assert_eq!(
+            state.events,
+            vec![
+                COMPRESS_DST_POLICY_OPEN,
+                COMPRESS_DST_POLICY_ATTACH,
+                COMPRESS_DST_POLICY_ADD_HANDLER,
+                COMPRESS_DST_POLICY_COMPRESS,
+                COMPRESS_DST_POLICY_CLEAR_HANDLER,
+                COMPRESS_DST_POLICY_CLOSE,
+            ]
+        );
+    }
+
    const DECOMPRESS_POLICY_OPEN_SOURCE: u8 = 1;
    const DECOMPRESS_POLICY_ASYNC: u8 = 2;
    const DECOMPRESS_POLICY_ATTACH_SOURCE: u8 = 3;