diff --git a/programs/fileio.c b/programs/fileio.c index eeb6aa1a2..60798d43f 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -404,6 +404,23 @@ void FIO_rust_freeDict(int dictBufferType, void** dictHandle); int FIO_rust_removeFile(const char* path); int FIO_rust_passThrough(ReadPoolCtx_t* readCtx, WritePoolCtx_t* writeCtx); +enum { + FIO_RUST_ZSTD_FRAME_OK = 0, + FIO_RUST_ZSTD_FRAME_DECODING_ERROR = 1, + FIO_RUST_ZSTD_FRAME_PREMATURE_END = 2, +}; +typedef void (*FIO_rust_frame_progress_fn)(void* opaque, + const char* srcFileName, + U64 decodedSize); +int FIO_rust_decompressZstdFrame(void* fCtx, + void* dctx, + ReadPoolCtx_t* readCtx, + WritePoolCtx_t* writeCtx, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* zstdError, + FIO_rust_frame_progress_fn progress); void FIO_rust_displayCompressionParameters(const FIO_prefs_t* prefs); #ifdef ZSTD_LZ4COMPRESS int FIO_rust_LZ4_GetBlockSize_FromBlockId(int id); @@ -2192,6 +2209,31 @@ FIO_zstdErrorHelp(const FIO_prefs_t* const prefs, * @return : size of decoded zstd frame, or an error code */ #define FIO_ERROR_FRAME_DECODING ((unsigned long long)(-2)) +static void +FIO_decompressZstdFrameProgress(void* const opaque, + const char* const srcFileName, + U64 const decodedSize) +{ + FIO_ctx_t* const fCtx = (FIO_ctx_t*)opaque; + const char* srcFName20 = srcFileName; + size_t const srcFileLength = strlen(srcFileName); + UTIL_HumanReadableSize_t const hrs = UTIL_makeHumanReadableSize(decodedSize); + + /* display last 20 characters only when not --verbose */ + if ((srcFileLength>20) && (g_display_prefs.displayLevel<3)) + srcFName20 += srcFileLength-20; + + if (fCtx->nbFilesTotal > 1) { + DISPLAYUPDATE_PROGRESS( + "\rDecompress: %2u/%2u files. Current: %s : %.*f%s... ", + fCtx->currFileIdx+1, fCtx->nbFilesTotal, srcFName20, + hrs.precision, hrs.value, hrs.suffix); + } else { + DISPLAYUPDATE_PROGRESS("\r%-20.20s : %.*f%s... ", + srcFName20, hrs.precision, hrs.value, hrs.suffix); + } +} + static unsigned long long FIO_decompressZstdFrame(FIO_ctx_t* const fCtx, dRess_t* ress, const FIO_prefs_t* const prefs, @@ -2199,68 +2241,28 @@ FIO_decompressZstdFrame(FIO_ctx_t* const fCtx, dRess_t* ress, U64 alreadyDecoded) /* for multi-frames streams */ { U64 frameSize = 0; - const char* srcFName20 = srcFileName; - IOJob_t* writeJob = AIO_WritePool_acquireJob(ress->writeCtx); - assert(writeJob); + size_t zstdError = 0; + int const status = FIO_rust_decompressZstdFrame( + fCtx, ress->dctx, ress->readCtx, ress->writeCtx, srcFileName, + alreadyDecoded, &frameSize, &zstdError, + FIO_decompressZstdFrameProgress); - /* display last 20 characters only when not --verbose */ - { size_t const srcFileLength = strlen(srcFileName); - if ((srcFileLength>20) && (g_display_prefs.displayLevel<3)) - srcFName20 += srcFileLength-20; + if (status == FIO_RUST_ZSTD_FRAME_OK) + return frameSize; + if (status == FIO_RUST_ZSTD_FRAME_DECODING_ERROR) { + DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n", + srcFileName, ZSTD_getErrorName(zstdError)); + FIO_zstdErrorHelp(prefs, ress, zstdError, srcFileName); + return FIO_ERROR_FRAME_DECODING; + } + if (status == FIO_RUST_ZSTD_FRAME_PREMATURE_END) { + DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n", + srcFileName); + return FIO_ERROR_FRAME_DECODING; } - ZSTD_DCtx_reset(ress->dctx, ZSTD_reset_session_only); - - /* Header loading : ensures ZSTD_getFrameHeader() will succeed */ - AIO_ReadPool_fillBuffer(ress->readCtx, ZSTD_FRAMEHEADERSIZE_MAX); - - /* Main decompression Loop */ - while (1) { - ZSTD_inBuffer inBuff = setInBuffer( ress->readCtx->srcBuffer, ress->readCtx->srcBufferLoaded, 0 ); - ZSTD_outBuffer outBuff= setOutBuffer( writeJob->buffer, writeJob->bufferSize, 0 ); - size_t const readSizeHint = ZSTD_decompressStream(ress->dctx, &outBuff, &inBuff); - UTIL_HumanReadableSize_t const hrs = UTIL_makeHumanReadableSize(alreadyDecoded+frameSize); - if (ZSTD_isError(readSizeHint)) { - DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n", - srcFileName, ZSTD_getErrorName(readSizeHint)); - FIO_zstdErrorHelp(prefs, ress, readSizeHint, srcFileName); - AIO_WritePool_releaseIoJob(writeJob); - return FIO_ERROR_FRAME_DECODING; - } - - /* Write block */ - writeJob->usedBufferSize = outBuff.pos; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - frameSize += outBuff.pos; - if (fCtx->nbFilesTotal > 1) { - DISPLAYUPDATE_PROGRESS( - "\rDecompress: %2u/%2u files. Current: %s : %.*f%s... ", - fCtx->currFileIdx+1, fCtx->nbFilesTotal, srcFName20, hrs.precision, hrs.value, hrs.suffix); - } else { - DISPLAYUPDATE_PROGRESS("\r%-20.20s : %.*f%s... ", - srcFName20, hrs.precision, hrs.value, hrs.suffix); - } - - AIO_ReadPool_consumeBytes(ress->readCtx, inBuff.pos); - - if (readSizeHint == 0) break; /* end of frame */ - - /* Fill input buffer */ - { size_t const toDecode = MIN(readSizeHint, ZSTD_DStreamInSize()); /* support large skippable frames */ - if (ress->readCtx->srcBufferLoaded < toDecode) { - size_t const readSize = AIO_ReadPool_fillBuffer(ress->readCtx, toDecode); - if (readSize==0) { - DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n", - srcFileName); - AIO_WritePool_releaseIoJob(writeJob); - return FIO_ERROR_FRAME_DECODING; - } - } } } - - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - - return frameSize; + assert(0); + return FIO_ERROR_FRAME_DECODING; } diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 5929a3cc8..920d9e682 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -1,6 +1,7 @@ #![allow(non_camel_case_types)] #![allow(non_snake_case)] #![allow(clippy::missing_safety_doc)] +#![allow(clippy::too_many_arguments)] //! Rust implementation of the exported asynchronous file-I/O pools. //! @@ -16,7 +17,7 @@ //! program, library, and C-test archives. use std::collections::VecDeque; -use std::ffi::c_void; +use std::ffi::{c_char, c_void}; use std::mem::{align_of, size_of}; #[cfg(any(windows, not(any(unix, windows))))] use std::os::raw::c_long; @@ -30,6 +31,20 @@ const IO_QUEUE_SIZE: usize = MAX_IO_JOBS - 2; const SPARSE_SEGMENT_SIZE: usize = 32 * 1024; const SPARSE_SKIP_CHUNK: u64 = 1 << 30; const PASS_THROUGH_MAX_BLOCK_SIZE: usize = 64 * 1024; +const ZSTD_FRAMEHEADERSIZE_MAX: usize = 18; + +pub const FIO_RUST_ZSTD_FRAME_OK: c_int = 0; +pub const FIO_RUST_ZSTD_FRAME_DECODING_ERROR: c_int = 1; +pub const FIO_RUST_ZSTD_FRAME_PREMATURE_END: c_int = 2; + +type FIO_rust_frame_progress_fn = Option; +type FIO_zstd_reset_fn = unsafe extern "C" fn(*mut c_void, c_int) -> usize; +type FIO_zstd_decompress_fn = unsafe extern "C" fn( + *mut c_void, + *mut crate::zstd_decompress::ZSTD_outBuffer, + *mut crate::zstd_decompress::ZSTD_inBuffer, +) -> usize; +type FIO_zstd_in_size_fn = extern "C" fn() -> usize; /// C's `FIO_prefs_t` from `programs/fileio_types.h`. /// @@ -1393,6 +1408,168 @@ pub unsafe extern "C" fn FIO_rust_passThrough( 0 } +/// Decompresses exactly one zstd frame using the existing asynchronous pools. +/// +/// C retains the frame dispatcher and all user-facing diagnostics. This ABI +/// deliberately exposes only opaque contexts, scalar counters, and output +/// parameters so the C-owned `dRess_t` never crosses into Rust. The input +/// pool is advanced only after a successful decoder call; in particular, an +/// error leaves the current input buffer untouched for `FIO_zstdErrorHelp()`. +unsafe fn decompress_zstd_frame_with( + f_ctx: *mut c_void, + dctx: *mut c_void, + read_ctx: *mut ReadPoolCtx_t, + write_ctx: *mut WritePoolCtx_t, + src_file_name: *const c_char, + already_decoded: u64, + frame_size: *mut u64, + zstd_error: *mut usize, + progress: FIO_rust_frame_progress_fn, + reset: FIO_zstd_reset_fn, + decompress: FIO_zstd_decompress_fn, + dstream_in_size: FIO_zstd_in_size_fn, +) -> c_int { + assert!(!dctx.is_null()); + assert!(!read_ctx.is_null()); + assert!(!write_ctx.is_null()); + assert!(!src_file_name.is_null()); + assert!(!frame_size.is_null()); + assert!(!zstd_error.is_null()); + + unsafe { + *frame_size = 0; + *zstd_error = 0; + } + + let mut write_job = unsafe { AIO_WritePool_acquireJob(write_ctx) }; + assert!(!write_job.is_null()); + + unsafe { + // The C implementation intentionally ignores this reset result. + reset(dctx, 1); // ZSTD_reset_session_only + AIO_ReadPool_fillBuffer(read_ctx, ZSTD_FRAMEHEADERSIZE_MAX); + } + + loop { + let mut input = crate::zstd_decompress::ZSTD_inBuffer { + src: unsafe { read_buffer_ptr(read_ctx) }.cast::(), + size: unsafe { read_buffer_loaded(read_ctx) }, + pos: 0, + }; + let mut output = crate::zstd_decompress::ZSTD_outBuffer { + dst: unsafe { (*write_job).buffer }, + size: unsafe { (*write_job).bufferSize }, + pos: 0, + }; + let read_size_hint = unsafe { decompress(dctx, &mut output, &mut input) }; + + if crate::errors::ERR_isError(read_size_hint) { + unsafe { + *zstd_error = read_size_hint; + AIO_WritePool_releaseIoJob(write_job); + } + return FIO_RUST_ZSTD_FRAME_DECODING_ERROR; + } + + let progress_size = unsafe { already_decoded.wrapping_add(*frame_size) }; + unsafe { + (*write_job).usedBufferSize = output.pos; + AIO_WritePool_enqueueAndReacquireWriteJob(&mut write_job); + *frame_size = (*frame_size).wrapping_add(output.pos as u64); + AIO_ReadPool_consumeBytes(read_ctx, input.pos); + } + + if let Some(callback) = progress { + unsafe { callback(f_ctx, src_file_name, progress_size) }; + } + + if read_size_hint == 0 { + unsafe { + AIO_WritePool_releaseIoJob(write_job); + AIO_WritePool_sparseWriteEnd(write_ctx); + } + return FIO_RUST_ZSTD_FRAME_OK; + } + + let to_decode = read_size_hint.min(dstream_in_size()); + let loaded = unsafe { read_buffer_loaded(read_ctx) }; + if loaded < to_decode { + let read_size = unsafe { AIO_ReadPool_fillBuffer(read_ctx, to_decode) }; + if read_size == 0 { + unsafe { AIO_WritePool_releaseIoJob(write_job) }; + return FIO_RUST_ZSTD_FRAME_PREMATURE_END; + } + } + } +} + +unsafe extern "C" fn fio_zstd_reset(dctx: *mut c_void, reset: c_int) -> usize { + unsafe { + crate::zstd_decompress::ZSTD_DCtx_reset( + dctx.cast::(), + reset, + ) + } +} + +unsafe extern "C" fn fio_zstd_decompress( + dctx: *mut c_void, + output: *mut crate::zstd_decompress::ZSTD_outBuffer, + input: *mut crate::zstd_decompress::ZSTD_inBuffer, +) -> usize { + unsafe { + crate::zstd_decompress::ZSTD_decompressStream( + dctx.cast::(), + output, + input, + ) + } +} + +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_decompressZstdFrame( + f_ctx: *mut c_void, + dctx: *mut c_void, + read_ctx: *mut ReadPoolCtx_t, + write_ctx: *mut WritePoolCtx_t, + src_file_name: *const c_char, + already_decoded: u64, + frame_size: *mut u64, + zstd_error: *mut usize, + progress: FIO_rust_frame_progress_fn, +) -> c_int { + unsafe { + decompress_zstd_frame_with( + f_ctx, + dctx, + read_ctx, + write_ctx, + src_file_name, + already_decoded, + frame_size, + zstd_error, + progress, + fio_zstd_reset, + fio_zstd_decompress, + crate::zstd_decompress::ZSTD_DStreamInSize, + ) + } +} + +#[inline] +unsafe fn read_buffer_ptr(ctx: *mut ReadPoolCtx_t) -> *const u8 { + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + unsafe { read_at::<*const u8>(context, inner.layout.read_src_buffer) } +} + +#[inline] +unsafe fn read_buffer_loaded(ctx: *mut ReadPoolCtx_t) -> usize { + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) } +} + impl PoolInner { unsafe fn all_jobs_available(&self) -> bool { let state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); @@ -1851,4 +2028,343 @@ mod tests { assert!(run_pass_through(&[], 5, 3, async_io).is_empty()); } } + + #[cfg(all(unix, feature = "compression"))] + mod zstd_frame_tests { + use super::*; + + #[repr(C)] + #[derive(Default)] + struct ProgressState { + calls: usize, + last_decoded: u64, + } + + unsafe extern "C" fn record_progress( + opaque: *mut c_void, + source_name: *const c_char, + decoded_size: u64, + ) { + assert!(!opaque.is_null()); + assert!(!source_name.is_null()); + let state = unsafe { &mut *opaque.cast::() }; + state.calls += 1; + state.last_decoded = decoded_size; + } + + const MOCK_FRAME_HEADER_SIZE: usize = 5; + const MOCK_DSTREAM_IN_SIZE: usize = 8; + const MOCK_FRAME_MAGIC: u8 = 0xA5; + + struct MockDecoder { + header: [u8; MOCK_FRAME_HEADER_SIZE], + header_len: usize, + remaining: usize, + } + + impl MockDecoder { + fn new() -> Self { + Self { + header: [0; MOCK_FRAME_HEADER_SIZE], + header_len: 0, + remaining: 0, + } + } + + fn reset(&mut self) { + self.header = [0; MOCK_FRAME_HEADER_SIZE]; + self.header_len = 0; + self.remaining = 0; + } + } + + fn encode_frame(input: &[u8]) -> Vec { + assert!(u32::try_from(input.len()).is_ok()); + let mut frame = Vec::with_capacity(MOCK_FRAME_HEADER_SIZE + input.len()); + frame.push(MOCK_FRAME_MAGIC); + frame.extend_from_slice(&(input.len() as u32).to_le_bytes()); + frame.extend_from_slice(input); + frame + } + + unsafe extern "C" fn mock_reset(dctx: *mut c_void, _reset: c_int) -> usize { + unsafe { (*dctx.cast::()).reset() }; + 0 + } + + unsafe extern "C" fn mock_decompress( + dctx: *mut c_void, + output: *mut crate::zstd_decompress::ZSTD_outBuffer, + input: *mut crate::zstd_decompress::ZSTD_inBuffer, + ) -> usize { + let decoder = unsafe { &mut *dctx.cast::() }; + let input = unsafe { &mut *input }; + let output = unsafe { &mut *output }; + let call_start = input.pos; + + while decoder.header_len < MOCK_FRAME_HEADER_SIZE && input.pos < input.size { + decoder.header[decoder.header_len] = + unsafe { input.src.cast::().add(input.pos).read() }; + decoder.header_len += 1; + input.pos += 1; + } + if decoder.header_len < MOCK_FRAME_HEADER_SIZE { + return (MOCK_FRAME_HEADER_SIZE - decoder.header_len).max(1); + } + if decoder.header[0] != MOCK_FRAME_MAGIC { + input.pos = call_start; + return crate::errors::ERROR(crate::errors::ZstdErrorCode::Generic); + } + if decoder.remaining == 0 { + let length = u32::from_le_bytes([ + decoder.header[1], + decoder.header[2], + decoder.header[3], + decoder.header[4], + ]); + decoder.remaining = length as usize; + if decoder.remaining == 0 { + return 0; + } + } + + let available = input.size - input.pos; + let writable = output.size - output.pos; + let copied = decoder.remaining.min(available).min(writable); + if copied != 0 { + unsafe { + ptr::copy_nonoverlapping( + input.src.cast::().add(input.pos), + output.dst.cast::().add(output.pos), + copied, + ); + } + input.pos += copied; + output.pos += copied; + decoder.remaining -= copied; + } + if decoder.remaining == 0 { + 0 + } else { + decoder.remaining.min(MOCK_DSTREAM_IN_SIZE) + } + } + + extern "C" fn mock_dstream_in_size() -> usize { + MOCK_DSTREAM_IN_SIZE + } + + struct FrameHarness { + _prefs: Box, + _decoder: Box, + read_ctx: *mut ReadPoolCtx_t, + write_ctx: *mut WritePoolCtx_t, + dctx: *mut c_void, + output_file: *mut libc::FILE, + } + + impl FrameHarness { + fn new(input: &[u8], read_buffer_size: usize, async_io: c_int) -> Self { + let mut prefs = Box::new(test_prefs(async_io)); + prefs.testMode = 0; + prefs.sparseFileSupport = 0; + + let input_file = unsafe { libc::tmpfile() }; + assert!(!input_file.is_null()); + assert_eq!( + unsafe { libc::fwrite(input.as_ptr().cast(), 1, input.len(), input_file) }, + input.len() + ); + assert_eq!(unsafe { libc::fflush(input_file) }, 0); + assert_eq!(unsafe { libc::fseek(input_file, 0, libc::SEEK_SET) }, 0); + + let output_file = unsafe { libc::tmpfile() }; + assert!(!output_file.is_null()); + let read_ctx = unsafe { AIO_ReadPool_create(&*prefs, read_buffer_size) }; + let write_ctx = unsafe { AIO_WritePool_create(&*prefs, 128 * 1024) }; + assert!(!read_ctx.is_null()); + assert!(!write_ctx.is_null()); + unsafe { + AIO_ReadPool_setFile(read_ctx, input_file); + AIO_WritePool_setFile(write_ctx, output_file); + } + let mut decoder = Box::new(MockDecoder::new()); + let dctx = (&mut *decoder as *mut MockDecoder).cast::(); + + Self { + _prefs: prefs, + _decoder: decoder, + read_ctx, + write_ctx, + dctx, + output_file, + } + } + + fn decompress( + &mut self, + already_decoded: u64, + progress_state: Option<&mut ProgressState>, + ) -> (c_int, u64, usize) { + let mut frame_size = 0; + let mut zstd_error = 0; + let progress_context = progress_state + .map(|state| state as *mut ProgressState as *mut c_void) + .unwrap_or(ptr::null_mut()); + let progress_callback = if progress_context.is_null() { + None + } else { + Some(record_progress as unsafe extern "C" fn(*mut c_void, *const c_char, u64)) + }; + let status = unsafe { + decompress_zstd_frame_with( + progress_context, + self.dctx, + self.read_ctx, + self.write_ctx, + c"frame-test.zst".as_ptr(), + already_decoded, + &mut frame_size, + &mut zstd_error, + progress_callback, + mock_reset, + mock_decompress, + mock_dstream_in_size, + ) + }; + (status, frame_size, zstd_error) + } + + fn unread_bytes(&self) -> Vec { + let loaded = unsafe { read_buffer_loaded(self.read_ctx) }; + let source = unsafe { read_buffer_ptr(self.read_ctx) }; + unsafe { std::slice::from_raw_parts(source, loaded) }.to_vec() + } + + fn output_bytes(&self) -> Vec { + assert_eq!( + unsafe { libc::fseek(self.output_file, 0, libc::SEEK_END) }, + 0 + ); + let size = unsafe { libc::ftell(self.output_file) }; + assert!(size >= 0); + let size = usize::try_from(size).unwrap(); + assert_eq!( + unsafe { libc::fseek(self.output_file, 0, libc::SEEK_SET) }, + 0 + ); + let mut output = vec![0_u8; size]; + if size != 0 { + assert_eq!( + unsafe { + libc::fread(output.as_mut_ptr().cast(), 1, size, self.output_file) + }, + size + ); + } + output + } + } + + impl Drop for FrameHarness { + fn drop(&mut self) { + unsafe { + AIO_WritePool_free(self.write_ctx); + AIO_ReadPool_free(self.read_ctx); + } + } + } + + #[test] + fn decompresses_frame_streams_output_and_reports_progress() { + let input: Vec = (0..300_000) + .map(|index| (index as u32).wrapping_mul(37).rotate_left(5) as u8) + .collect(); + let frame = encode_frame(&input); + + for async_io in [0, 1] { + let mut harness = FrameHarness::new(&frame, 97, async_io); + let mut progress = ProgressState::default(); + let (status, frame_size, zstd_error) = harness.decompress(0, Some(&mut progress)); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK); + assert_eq!(frame_size, input.len() as u64); + assert_eq!(zstd_error, 0); + assert!(progress.calls > 1); + assert!(progress.last_decoded < input.len() as u64); + assert_eq!(harness.output_bytes(), input); + } + } + + #[test] + fn progress_uses_the_pre_increment_frame_size() { + let input = b"one streaming output block"; + let frame = encode_frame(input); + let mut harness = FrameHarness::new(&frame, frame.len() + 1, 0); + let mut progress = ProgressState::default(); + + let (status, frame_size, zstd_error) = harness.decompress(37, Some(&mut progress)); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK); + assert_eq!(frame_size, input.len() as u64); + assert_eq!(zstd_error, 0); + assert_eq!(progress.calls, 1); + assert_eq!(progress.last_decoded, 37); + } + + #[test] + fn decompresses_concatenated_frames_without_crossing_boundary() { + let first = b"first frame ".repeat(20_000); + let second = b"second frame ".repeat(20_000); + let first_frame = encode_frame(&first); + let second_frame = encode_frame(&second); + let mut stream = first_frame.clone(); + stream.extend_from_slice(&second_frame); + + let mut harness = FrameHarness::new(&stream, stream.len(), 1); + let mut progress = ProgressState::default(); + let (status, frame_size, zstd_error) = harness.decompress(0, Some(&mut progress)); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK); + assert_eq!(frame_size, first.len() as u64); + assert_eq!(zstd_error, 0); + assert_eq!(harness.unread_bytes(), second_frame); + assert_eq!(harness.output_bytes(), first); + + let (status, frame_size, zstd_error) = + harness.decompress(first.len() as u64, Some(&mut progress)); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK); + assert_eq!(frame_size, second.len() as u64); + assert_eq!(zstd_error, 0); + assert!(harness.unread_bytes().is_empty()); + assert!(progress.last_decoded >= first.len() as u64); + assert!(progress.last_decoded < (first.len() + second.len()) as u64); + + let mut expected = first; + expected.extend_from_slice(&second); + assert_eq!(harness.output_bytes(), expected); + } + + #[test] + fn decoder_error_preserves_the_current_input_buffer() { + let input = b"decoder error input"; + let mut frame = encode_frame(input); + frame[0] ^= 1; + let mut harness = FrameHarness::new(&frame, frame.len() + 1, 0); + + let (status, frame_size, zstd_error) = harness.decompress(0, None); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_DECODING_ERROR); + assert_eq!(frame_size, 0); + assert_ne!(zstd_error, 0); + assert_eq!(harness.unread_bytes(), frame); + } + + #[test] + fn truncated_frame_reports_premature_end() { + let frame = encode_frame(&b"truncated frame".repeat(10_000)); + let truncated = &frame[..frame.len() - 1]; + let mut harness = FrameHarness::new(truncated, truncated.len() + 1, 0); + + let (status, _frame_size, zstd_error) = harness.decompress(0, None); + assert_eq!(status, FIO_RUST_ZSTD_FRAME_PREMATURE_END); + assert_eq!(zstd_error, 0); + } + } }