From 7af1b2714006ff0ed6efc593882e6863a159a04a Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 02:34:59 +0200 Subject: [PATCH] feat(fileio): move source compression policy into Rust MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the ordered policy from FIO_compressFilename_srcFile into a narrow Rust projection: named-source stat and collision checks, compressed-file exclusion, source open and close, size-based async selection, source attachment, and successful-only --rm handling now run in Rust. The destination callback still hands compression to the existing Rust compression loop. Keep FIO_ctx_t, cRess_t, stat_t, FILE/pool handles, signal handling, and CLI diagnostics behind opaque C callbacks, so the ABI carries only filenames, scalar policy flags, and callback pointers. Focused Rust policy tests cover named-file short circuits, stdin behavior, async threshold selection, and the successful-compression removal gate. Test Plan: - `rustfmt --edition 2021 --check rust/src/fileio_asyncio.rs` — passed - `git diff --check` and path-limited `git diff --cached --check` — passed - Static ABI/order/diff inspection only; policy tests were not executed - cargo/make/native/fuzzer/heavy tests were not run per request --- programs/fileio.c | 255 +++++++++++++++++------ rust/src/fileio_asyncio.rs | 412 +++++++++++++++++++++++++++++++++++++ 2 files changed, 605 insertions(+), 62 deletions(-) diff --git a/programs/fileio.c b/programs/fileio.c index f86769747..8e8b21802 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1460,6 +1460,76 @@ int FIO_rust_compressFilenameInternal( U64 fileSize, int compressionLevel, const FIO_rust_compress_callbacks_t* callbacks); +enum { + FIO_RUST_COMPRESS_SRC_STAT_FAILED = 0, + FIO_RUST_COMPRESS_SRC_STAT_OK = 1, + FIO_RUST_COMPRESS_SRC_DIRECTORY = 2, + FIO_RUST_COMPRESS_SRC_DICT_COLLISION = 3, + FIO_RUST_COMPRESS_SRC_OPEN_OK = 0, +}; +typedef int (*FIO_rust_compress_src_stat_fn)(void* opaque, const char* srcFileName); +typedef int (*FIO_rust_compress_src_excluded_fn)(void* opaque, const char* srcFileName); +typedef int (*FIO_rust_compress_src_open_fn)(void* opaque, const char* srcFileName, + U64* fileSize); +typedef void (*FIO_rust_compress_src_async_fn)(void* opaque, int asyncMode); +typedef void (*FIO_rust_compress_src_attach_fn)(void* opaque); +typedef int (*FIO_rust_compress_src_compress_fn)(void* opaque, + const char* dstFileName, + const char* srcFileName, + int compressionLevel); +typedef void (*FIO_rust_compress_src_close_fn)(void* opaque); +typedef void (*FIO_rust_compress_src_remove_fn)(void* opaque, const char* srcFileName); + +/* Rust owns the source policy; this projection exposes only scalar names and + * opaque C callbacks. FIO_ctx_t, cRess_t, stat_t, and all resource/diagnostic + * state remain private to the callback context below. */ +typedef struct { + void* opaque; + const char* dstFileName; + const char* srcFileName; + int compressionLevel; + int sourceIsStdin; + int excludeCompressedFiles; + int removeSrcFile; + FIO_rust_compress_src_stat_fn statSource; + FIO_rust_compress_src_excluded_fn sourceIsExcluded; + FIO_rust_compress_src_open_fn openSource; + FIO_rust_compress_src_async_fn setAsync; + FIO_rust_compress_src_attach_fn attachSource; + FIO_rust_compress_src_compress_fn compress; + FIO_rust_compress_src_close_fn closeSource; + FIO_rust_compress_src_remove_fn removeSource; +} FIO_rust_compress_src_projection_t; +typedef char FIO_rust_compress_src_opaque_offset[ + (offsetof(FIO_rust_compress_src_projection_t, opaque) == 0) ? 1 : -1]; +typedef char FIO_rust_compress_src_dst_name_offset[ + (offsetof(FIO_rust_compress_src_projection_t, dstFileName) + == sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_src_src_name_offset[ + (offsetof(FIO_rust_compress_src_projection_t, srcFileName) + == 2 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_src_level_offset[ + (offsetof(FIO_rust_compress_src_projection_t, compressionLevel) + == 3 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_src_stdin_offset[ + (offsetof(FIO_rust_compress_src_projection_t, sourceIsStdin) + == 3 * sizeof(void*) + sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_src_exclude_offset[ + (offsetof(FIO_rust_compress_src_projection_t, excludeCompressedFiles) + == 3 * sizeof(void*) + 2 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_src_remove_offset[ + (offsetof(FIO_rust_compress_src_projection_t, removeSrcFile) + == 3 * sizeof(void*) + 3 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_src_stat_callback_offset[ + (offsetof(FIO_rust_compress_src_projection_t, statSource) + == 3 * sizeof(void*) + 4 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_src_projection_size[ + (sizeof(FIO_rust_compress_src_projection_t) + == 3 * sizeof(void*) + 4 * sizeof(int) + + 8 * sizeof(FIO_rust_compress_src_stat_fn)) ? 1 : -1]; +int FIO_rust_compressFilenameSrcFile( + const FIO_rust_compress_src_projection_t* projection); + static void FIO_adjustParamsForPatchFromMode(FIO_prefs_t* const prefs, ZSTD_compressionParameters* comprParams, unsigned long long const dictSize, @@ -2891,6 +2961,105 @@ static const char *compressedFileExtensions[] = { NULL }; +typedef struct { + FIO_ctx_t* fCtx; + FIO_prefs_t* prefs; + cRess_t* ress; + stat_t srcFileStat; + FILE* srcFile; +} FIO_rust_compress_src_context_t; + +static int FIO_rust_compressSourceStat(void* opaque, const char* srcFileName) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + + if (!UTIL_stat(srcFileName, &context->srcFileStat)) { + /* Failure to stat at all is handled during opening. */ + return FIO_RUST_COMPRESS_SRC_STAT_FAILED; + } + + if (UTIL_isDirectoryStat(&context->srcFileStat)) { + DISPLAYLEVEL(1, "zstd: %s is a directory -- ignored \n", srcFileName); + return FIO_RUST_COMPRESS_SRC_DIRECTORY; + } + + if (context->ress->dictFileName != NULL + && UTIL_isSameFileStat(srcFileName, context->ress->dictFileName, + &context->srcFileStat, &context->ress->dictFileStat)) { + DISPLAYLEVEL(1, "zstd: cannot use %s as an input file and dictionary \n", srcFileName); + return FIO_RUST_COMPRESS_SRC_DICT_COLLISION; + } + + return FIO_RUST_COMPRESS_SRC_STAT_OK; +} + +static int FIO_rust_compressSourceIsExcluded(void* opaque, const char* srcFileName) +{ + int const isCompressed = UTIL_isCompressedFile(srcFileName, compressedFileExtensions); + (void)opaque; + if (isCompressed) + DISPLAYLEVEL(4, "File is already compressed : %s \n", srcFileName); + return isCompressed; +} + +static int FIO_rust_compressSourceOpen(void* opaque, const char* srcFileName, + U64* fileSize) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + context->srcFile = FIO_openSrcFile(context->prefs, srcFileName, &context->srcFileStat); + if (context->srcFile == NULL) + return 1; + + *fileSize = strcmp(srcFileName, stdinmark) + ? UTIL_getFileSizeStat(&context->srcFileStat) + : UTIL_FILESIZE_UNKNOWN; + return FIO_RUST_COMPRESS_SRC_OPEN_OK; +} + +static void FIO_rust_compressSourceSetAsync(void* opaque, int asyncMode) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + AIO_ReadPool_setAsync(context->ress->readCtx, asyncMode); + AIO_WritePool_setAsync(context->ress->writeCtx, asyncMode); +} + +static void FIO_rust_compressSourceAttach(void* opaque) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + AIO_ReadPool_setFile(context->ress->readCtx, context->srcFile); +} + +static int FIO_rust_compressSourceCompress(void* opaque, const char* dstFileName, + const char* srcFileName, int compressionLevel) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + return FIO_compressFilename_dstFile( + context->fCtx, context->prefs, *context->ress, + dstFileName, srcFileName, &context->srcFileStat, compressionLevel); +} + +static void FIO_rust_compressSourceClose(void* opaque) +{ + FIO_rust_compress_src_context_t* const context = + (FIO_rust_compress_src_context_t*)opaque; + (void)AIO_ReadPool_closeFile(context->ress->readCtx); + context->srcFile = NULL; +} + +static void FIO_rust_compressSourceRemove(void* opaque, const char* srcFileName) +{ + (void)opaque; + /* After this point Ctrl-C must not remove both source and destination. */ + clearHandler(); + if (FIO_removeFile(srcFileName)) + EXM_THROW(1, "zstd: %s: %s", srcFileName, strerror(errno)); +} + /*! FIO_compressFilename_srcFile() : * @return : 0 : compression completed correctly, * 1 : missing or pb opening srcFileName @@ -2903,72 +3072,34 @@ FIO_compressFilename_srcFile(FIO_ctx_t* const fCtx, const char* srcFileName, int compressionLevel) { - int result; - FILE* srcFile; - stat_t srcFileStat; - U64 fileSize = UTIL_FILESIZE_UNKNOWN; + FIO_rust_compress_src_context_t context; + FIO_rust_compress_src_projection_t projection; + DISPLAYLEVEL(6, "FIO_compressFilename_srcFile: %s \n", srcFileName); - if (strcmp(srcFileName, stdinmark)) { - if (UTIL_stat(srcFileName, &srcFileStat)) { - /* failure to stat at all is handled during opening */ + memset(&context, 0, sizeof(context)); + context.fCtx = fCtx; + context.prefs = prefs; + context.ress = &ress; - /* ensure src is not a directory */ - if (UTIL_isDirectoryStat(&srcFileStat)) { - DISPLAYLEVEL(1, "zstd: %s is a directory -- ignored \n", srcFileName); - return 1; - } + memset(&projection, 0, sizeof(projection)); + projection.opaque = &context; + projection.dstFileName = dstFileName; + projection.srcFileName = srcFileName; + projection.compressionLevel = compressionLevel; + projection.sourceIsStdin = !strcmp(srcFileName, stdinmark); + projection.excludeCompressedFiles = prefs->excludeCompressedFiles == 1; + projection.removeSrcFile = prefs->removeSrcFile; + projection.statSource = FIO_rust_compressSourceStat; + projection.sourceIsExcluded = FIO_rust_compressSourceIsExcluded; + projection.openSource = FIO_rust_compressSourceOpen; + projection.setAsync = FIO_rust_compressSourceSetAsync; + projection.attachSource = FIO_rust_compressSourceAttach; + projection.compress = FIO_rust_compressSourceCompress; + projection.closeSource = FIO_rust_compressSourceClose; + projection.removeSource = FIO_rust_compressSourceRemove; - /* ensure src is not the same as dict (if present) */ - if (ress.dictFileName != NULL && UTIL_isSameFileStat(srcFileName, ress.dictFileName, &srcFileStat, &ress.dictFileStat)) { - DISPLAYLEVEL(1, "zstd: cannot use %s as an input file and dictionary \n", srcFileName); - return 1; - } - } - } - - /* Check if "srcFile" is compressed. Only done if --exclude-compressed flag is used - * YES => ZSTD will skip compression of the file and will return 0. - * NO => ZSTD will resume with compress operation. - */ - if (prefs->excludeCompressedFiles == 1 && UTIL_isCompressedFile(srcFileName, compressedFileExtensions)) { - DISPLAYLEVEL(4, "File is already compressed : %s \n", srcFileName); - return 0; - } - - srcFile = FIO_openSrcFile(prefs, srcFileName, &srcFileStat); - if (srcFile == NULL) return 1; /* srcFile could not be opened */ - - /* Don't use AsyncIO for small files */ - if (strcmp(srcFileName, stdinmark)) /* Stdin doesn't have stats */ - fileSize = UTIL_getFileSizeStat(&srcFileStat); - if(fileSize != UTIL_FILESIZE_UNKNOWN && fileSize < ZSTD_BLOCKSIZE_MAX * 3) { - AIO_ReadPool_setAsync(ress.readCtx, 0); - AIO_WritePool_setAsync(ress.writeCtx, 0); - } else { - AIO_ReadPool_setAsync(ress.readCtx, 1); - AIO_WritePool_setAsync(ress.writeCtx, 1); - } - - AIO_ReadPool_setFile(ress.readCtx, srcFile); - result = FIO_compressFilename_dstFile( - fCtx, prefs, ress, - dstFileName, srcFileName, - &srcFileStat, compressionLevel); - AIO_ReadPool_closeFile(ress.readCtx); - - if ( prefs->removeSrcFile /* --rm */ - && result == 0 /* success */ - && strcmp(srcFileName, stdinmark) /* exception : don't erase stdin */ - ) { - /* We must clear the handler, since after this point calling it would - * delete both the source and destination files. - */ - clearHandler(); - if (FIO_removeFile(srcFileName)) - EXM_THROW(1, "zstd: %s: %s", srcFileName, strerror(errno)); - } - return result; + return FIO_rust_compressFilenameSrcFile(&projection); } void FIO_displayCompressionParameters(const FIO_prefs_t* prefs) diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 66d272076..2f1faac2a 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -259,6 +259,112 @@ pub struct FIO_rust_compress_callbacks_t { pub display_status: Option, } +pub const FIO_RUST_COMPRESS_SRC_STAT_FAILED: c_int = 0; +pub const FIO_RUST_COMPRESS_SRC_STAT_OK: c_int = 1; +pub const FIO_RUST_COMPRESS_SRC_DIRECTORY: c_int = 2; +pub const FIO_RUST_COMPRESS_SRC_DICT_COLLISION: c_int = 3; + +const FIO_RUST_COMPRESS_SRC_OPEN_OK: c_int = 0; +const FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD: u64 = (1 << 17) * 3; +const FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE: u64 = u64::MAX; + +pub type FIO_rust_compress_src_stat_fn = unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int; +pub type FIO_rust_compress_src_excluded_fn = + unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int; +pub type FIO_rust_compress_src_open_fn = + unsafe extern "C" fn(*mut c_void, *const c_char, *mut u64) -> c_int; +pub type FIO_rust_compress_src_async_fn = unsafe extern "C" fn(*mut c_void, c_int); +pub type FIO_rust_compress_src_attach_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_compress_src_compress_fn = + unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int) -> c_int; +pub type FIO_rust_compress_src_close_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_compress_src_remove_fn = unsafe extern "C" fn(*mut c_void, *const c_char); + +/// Rust owns the source-file policy and ordering. C retains the private +/// context/resource/stat objects and implements each operation behind opaque +/// callbacks, including diagnostics, signal handling, and actual compression. +#[repr(C)] +pub struct FIO_rust_compress_src_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 source_is_stdin: c_int, + pub exclude_compressed_files: c_int, + pub remove_src_file: c_int, + pub stat_source: Option, + pub source_is_excluded: Option, + pub open_source: Option, + pub set_async: Option, + pub attach_source: Option, + pub compress: Option, + pub close_source: Option, + pub remove_source: Option, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_compress_src_projection_t, opaque) == 0); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, dst_file_name) + == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, src_file_name) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, compression_level) + == 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, source_is_stdin) + == 3 * size_of::() + size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, exclude_compressed_files) + == 3 * size_of::() + 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, remove_src_file) + == 3 * size_of::() + 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, stat_source) + == 3 * size_of::() + 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, source_is_excluded) + == 3 * size_of::() + 4 * size_of::() + size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, open_source) + == 3 * size_of::() + 4 * size_of::() + 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, set_async) + == 3 * size_of::() + 4 * size_of::() + 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, attach_source) + == 3 * size_of::() + 4 * size_of::() + 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, compress) + == 3 * size_of::() + 4 * size_of::() + 5 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, close_source) + == 3 * size_of::() + 4 * size_of::() + 6 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_src_projection_t, remove_source) + == 3 * size_of::() + 4 * size_of::() + 7 * size_of::() + ); + assert!( + size_of::() + == 3 * size_of::() + 4 * size_of::() + 8 * size_of::() + ); +}; + 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 = @@ -2267,6 +2373,96 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal( FIO_RUST_COMPRESS_OK } +/// Runs the source-file policy around the existing C-owned destination and +/// compression callback. Rust owns the ordering: named-file checks, +/// exclusion, source open, size-based async selection, source attachment, +/// compression, close, and finally conditional source removal. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_compressFilenameSrcFile( + projection: *const FIO_rust_compress_src_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()); + + let src_file_name = projection.src_file_name; + let source_is_stdin = projection.source_is_stdin != 0; + + if !source_is_stdin { + let stat_source = projection + .stat_source + .expect("source stat callback is required for named files"); + match unsafe { stat_source(projection.opaque, src_file_name) } { + FIO_RUST_COMPRESS_SRC_STAT_FAILED | FIO_RUST_COMPRESS_SRC_STAT_OK => {} + FIO_RUST_COMPRESS_SRC_DIRECTORY | FIO_RUST_COMPRESS_SRC_DICT_COLLISION => return 1, + _ => unreachable!("invalid source stat status"), + } + } + + if projection.exclude_compressed_files != 0 { + let source_is_excluded = projection + .source_is_excluded + .expect("compressed-file predicate is required when exclusion is enabled"); + if unsafe { source_is_excluded(projection.opaque, src_file_name) } != 0 { + return 0; + } + } + + let open_source = projection + .open_source + .expect("source open callback is required"); + let mut file_size = FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE; + let open_status = unsafe { open_source(projection.opaque, src_file_name, &mut file_size) }; + if open_status != FIO_RUST_COMPRESS_SRC_OPEN_OK { + return 1; + } + + let async_mode = if file_size != FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE + && file_size < FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD + { + 0 + } else { + 1 + }; + let set_async = projection + .set_async + .expect("async-selection callback is required"); + unsafe { set_async(projection.opaque, async_mode) }; + + let attach_source = projection + .attach_source + .expect("source attachment callback is required"); + unsafe { attach_source(projection.opaque) }; + + let compress = projection + .compress + .expect("compression callback is required"); + let result = unsafe { + compress( + projection.opaque, + projection.dst_file_name, + src_file_name, + projection.compression_level, + ) + }; + + let close_source = projection + .close_source + .expect("source close callback is required"); + unsafe { close_source(projection.opaque) }; + + if projection.remove_src_file != 0 && result == 0 && !source_is_stdin { + let remove_source = projection + .remove_source + .expect("source removal callback is required when removal is enabled"); + unsafe { remove_source(projection.opaque, src_file_name) }; + } + + result +} + /// 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 @@ -4277,6 +4473,222 @@ impl PoolInner { mod tests { use super::*; + const SOURCE_POLICY_STAT: u8 = 1; + const SOURCE_POLICY_EXCLUDED: u8 = 2; + const SOURCE_POLICY_OPEN: u8 = 3; + const SOURCE_POLICY_ASYNC: u8 = 4; + const SOURCE_POLICY_ATTACH: u8 = 5; + const SOURCE_POLICY_COMPRESS: u8 = 6; + const SOURCE_POLICY_CLOSE: u8 = 7; + const SOURCE_POLICY_REMOVE: u8 = 8; + + #[derive(Default)] + struct SourcePolicyState { + events: Vec, + async_modes: Vec, + stat_status: c_int, + excluded: c_int, + open_status: c_int, + file_size: u64, + compression_status: c_int, + compression_level: c_int, + } + + unsafe extern "C" fn source_policy_stat( + opaque: *mut c_void, + _src_file_name: *const c_char, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_STAT); + state.stat_status + } + + unsafe extern "C" fn source_policy_excluded( + opaque: *mut c_void, + _src_file_name: *const c_char, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_EXCLUDED); + state.excluded + } + + unsafe extern "C" fn source_policy_open( + opaque: *mut c_void, + _src_file_name: *const c_char, + file_size: *mut u64, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_OPEN); + unsafe { *file_size = state.file_size }; + state.open_status + } + + unsafe extern "C" fn source_policy_async(opaque: *mut c_void, async_mode: c_int) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_ASYNC); + state.async_modes.push(async_mode); + } + + unsafe extern "C" fn source_policy_attach(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_ATTACH); + } + + unsafe extern "C" fn source_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::() }; + state.events.push(SOURCE_POLICY_COMPRESS); + state.compression_level = compression_level; + state.compression_status + } + + unsafe extern "C" fn source_policy_close(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_CLOSE); + } + + unsafe extern "C" fn source_policy_remove(opaque: *mut c_void, _src_file_name: *const c_char) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(SOURCE_POLICY_REMOVE); + } + + fn source_policy_projection( + state: &mut SourcePolicyState, + source_is_stdin: c_int, + exclude_compressed_files: c_int, + remove_src_file: c_int, + ) -> FIO_rust_compress_src_projection_t { + FIO_rust_compress_src_projection_t { + opaque: (state as *mut SourcePolicyState).cast(), + dst_file_name: c"destination".as_ptr(), + src_file_name: c"source".as_ptr(), + compression_level: 7, + source_is_stdin, + exclude_compressed_files, + remove_src_file, + stat_source: Some(source_policy_stat), + source_is_excluded: Some(source_policy_excluded), + open_source: Some(source_policy_open), + set_async: Some(source_policy_async), + attach_source: Some(source_policy_attach), + compress: Some(source_policy_compress), + close_source: Some(source_policy_close), + remove_source: Some(source_policy_remove), + } + } + + #[test] + fn source_policy_keeps_named_checks_before_exclusion_and_open() { + for status in [ + FIO_RUST_COMPRESS_SRC_DIRECTORY, + FIO_RUST_COMPRESS_SRC_DICT_COLLISION, + ] { + let mut state = SourcePolicyState { + stat_status: status, + excluded: 1, + ..SourcePolicyState::default() + }; + let projection = source_policy_projection(&mut state, 0, 1, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 1); + assert_eq!(state.events, vec![SOURCE_POLICY_STAT]); + } + } + + #[test] + fn source_policy_short_circuits_excluded_files_before_opening() { + let mut state = SourcePolicyState { + stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK, + excluded: 1, + ..SourcePolicyState::default() + }; + let projection = source_policy_projection(&mut state, 0, 1, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0); + assert_eq!( + state.events, + vec![SOURCE_POLICY_STAT, SOURCE_POLICY_EXCLUDED] + ); + } + + #[test] + fn source_policy_selects_async_after_open_and_removes_only_after_success() { + let mut state = SourcePolicyState { + stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK, + file_size: FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD - 1, + compression_status: 0, + ..SourcePolicyState::default() + }; + let projection = source_policy_projection(&mut state, 0, 0, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0); + assert_eq!(state.async_modes, vec![0]); + assert_eq!(state.compression_level, 7); + assert_eq!( + state.events, + vec![ + SOURCE_POLICY_STAT, + SOURCE_POLICY_OPEN, + SOURCE_POLICY_ASYNC, + SOURCE_POLICY_ATTACH, + SOURCE_POLICY_COMPRESS, + SOURCE_POLICY_CLOSE, + SOURCE_POLICY_REMOVE, + ] + ); + + let mut failed = SourcePolicyState { + stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK, + file_size: FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD, + compression_status: 1, + ..SourcePolicyState::default() + }; + let failed_projection = source_policy_projection(&mut failed, 0, 0, 1); + + assert_eq!( + unsafe { FIO_rust_compressFilenameSrcFile(&failed_projection) }, + 1 + ); + assert_eq!(failed.async_modes, vec![1]); + assert_eq!( + failed.events, + vec![ + SOURCE_POLICY_STAT, + SOURCE_POLICY_OPEN, + SOURCE_POLICY_ASYNC, + SOURCE_POLICY_ATTACH, + SOURCE_POLICY_COMPRESS, + SOURCE_POLICY_CLOSE, + ] + ); + } + + #[test] + fn source_policy_skips_named_checks_and_removal_for_stdin() { + let mut state = SourcePolicyState { + file_size: FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE, + ..SourcePolicyState::default() + }; + let projection = source_policy_projection(&mut state, 1, 0, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0); + assert_eq!(state.async_modes, vec![1]); + assert_eq!( + state.events, + vec![ + SOURCE_POLICY_OPEN, + SOURCE_POLICY_ASYNC, + SOURCE_POLICY_ATTACH, + SOURCE_POLICY_COMPRESS, + SOURCE_POLICY_CLOSE, + ] + ); + } + fn run_zstd_adapt(policy: c_int, projection: &FIO_rust_zstd_adapt_projection_t) -> c_int { unsafe { FIO_rust_zstd_adapt(policy, projection) } }