mod common;
use common::*;
use hyperfile::file::hyper::Hyper;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::mode::FileMode;
use hyperfile::buffer::{AlignedDataBlockWrapper, BatchDataBlockWrapper};
#[tokio::test]
#[ignore]
async fn rollback_exposure_truncate_extend_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 1024usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = AlwaysFailFlushInode::new();
let truncate_result = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let result = hyper.fs_truncate(8192).await;
let _release_result = hyper.fs_release().await;
result
};
assert!(
truncate_result.is_err(),
"expected truncate to fail due to injected flush_inode failure, got {:?}",
truncate_result
);
assert!(
interceptor.call_count() >= 1,
"interceptor was never invoked"
);
println!(
"[truncate_extend_flush_fails] truncate_result={:?} interceptor_calls={}",
truncate_result,
interceptor.call_count()
);
let persisted_size = {
let hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let stat = hyper.fs_getattr().expect("getattr");
stat.st_size as usize
};
println!(
"[truncate_extend_flush_fails] persisted_size={}",
persisted_size
);
assert_eq!(
persisted_size, original_size,
"persisted size diverged: expected original {}, got {}",
original_size, persisted_size
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_exposure_truncate_shrink_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 16 * 1024usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = AlwaysFailFlushInode::new();
let truncate_result = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let result = hyper.fs_truncate(4096).await;
let _release = hyper.fs_release().await;
result
};
assert!(
truncate_result.is_err(),
"expected truncate to fail, got {:?}",
truncate_result
);
println!(
"[truncate_shrink_flush_fails] truncate_result={:?} interceptor_calls={}",
truncate_result,
interceptor.call_count()
);
let (persisted_size, first_block_ok, last_block_ok) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let stat = hyper.fs_getattr().expect("getattr");
let size = stat.st_size as usize;
let mut buf0 = vec![0u8; 4096];
let r0 = hyper.fs_read(0, &mut buf0).await;
let first_block_ok = r0.is_ok() && buf0 == payload[..4096];
let last_block_ok = if size >= 16 * 1024 {
let mut buf3 = vec![0u8; 4096];
let r3 = hyper.fs_read(12 * 1024, &mut buf3).await;
Some(r3.is_ok() && buf3 == payload[12 * 1024..16 * 1024])
} else {
None
};
(size, first_block_ok, last_block_ok)
};
println!(
"[truncate_shrink_flush_fails] persisted_size={} first_block_ok={} last_block_ok={:?}",
persisted_size, first_block_ok, last_block_ok
);
assert_eq!(
persisted_size, original_size,
"persisted size diverged after failed shrink: expected {}, got {}",
original_size, persisted_size
);
assert!(first_block_ok, "first block corrupted after failed shrink");
if let Some(ok) = last_block_ok {
assert!(ok, "last block unreadable after failed shrink — data lost");
}
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_exposure_write_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = AlwaysFailFlushInode::new();
let (write_result, flush_result) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let data = b"hello world";
let wr = hyper.fs_write(0, data).await;
let fl = hyper.fs_flush().await;
let _release = hyper.fs_release().await;
(wr, fl)
};
println!(
"[write_flush_fails] write_result={:?} flush_result={:?} interceptor_calls={}",
write_result,
flush_result,
interceptor.call_count()
);
assert!(
flush_result.is_err(),
"expected flush to fail, got write={:?} flush={:?}",
write_result,
flush_result
);
let persisted_size = {
let hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let stat = hyper.fs_getattr().expect("getattr");
stat.st_size as usize
};
println!(
"[write_flush_fails] persisted_size={}",
persisted_size
);
assert_eq!(
persisted_size, 0,
"persisted size diverged after failed write+flush: expected 0, got {}",
persisted_size
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_truncate_extend_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 1024usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let truncate_result = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let r = hyper.fs_truncate(8192).await;
let _rel = hyper.fs_release().await;
r
};
assert!(
truncate_result.is_err(),
"expected truncate to fail, got {:?}",
truncate_result
);
println!(
"[correctness_truncate_extend] truncate={:?} calls={}",
truncate_result,
interceptor.call_count()
);
let persisted_size = {
let hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
hyper.fs_getattr().expect("getattr").st_size as usize
};
println!(
"[correctness_truncate_extend] persisted_size={}",
persisted_size
);
assert_eq!(
persisted_size, original_size,
"failed truncate silently committed: expected size {}, got {}",
original_size, persisted_size
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_truncate_shrink_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 16 * 1024usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let truncate_result = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let r = hyper.fs_truncate(4096).await;
let _rel = hyper.fs_release().await;
r
};
assert!(
truncate_result.is_err(),
"expected truncate to fail, got {:?}",
truncate_result
);
println!(
"[correctness_truncate_shrink] truncate={:?} calls={}",
truncate_result,
interceptor.call_count()
);
let (persisted_size, first_ok, last_ok) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let size = hyper.fs_getattr().expect("getattr").st_size as usize;
let mut b0 = vec![0u8; 4096];
let r0 = hyper.fs_read(0, &mut b0).await;
let first_ok = r0.is_ok() && b0 == payload[..4096];
let last_ok = if size >= 16 * 1024 {
let mut b3 = vec![0u8; 4096];
let r3 = hyper.fs_read(12 * 1024, &mut b3).await;
Some(r3.is_ok() && b3 == payload[12 * 1024..16 * 1024])
} else {
None
};
(size, first_ok, last_ok)
};
println!(
"[correctness_truncate_shrink] size={} first_ok={} last_ok={:?}",
persisted_size, first_ok, last_ok
);
assert_eq!(
persisted_size, original_size,
"failed shrink silently committed: size expected {}, got {}",
original_size, persisted_size
);
assert!(first_ok, "first block corrupted after failed shrink");
assert_eq!(
last_ok,
Some(true),
"last block unreadable after failed shrink — data lost"
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_write_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let (write_result, flush_result) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let wr = hyper.fs_write(0, b"hello world").await;
let fl = hyper.fs_flush().await;
let _rel = hyper.fs_release().await;
(wr, fl)
};
println!(
"[correctness_write] write={:?} flush={:?} calls={}",
write_result,
flush_result,
interceptor.call_count()
);
assert!(
flush_result.is_err(),
"expected flush to fail, got write={:?} flush={:?}",
write_result,
flush_result
);
let persisted_size = {
let hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
hyper.fs_getattr().expect("getattr").st_size as usize
};
println!("[correctness_write] persisted_size={}", persisted_size);
assert_eq!(
persisted_size, 0,
"failed write silently committed: expected 0, got {}",
persisted_size
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_write_zero_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 4096usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let (write_zero_result, flush_result) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let wr = hyper.fs_write_zero(original_size, 8192).await;
let fr = hyper.fs_flush().await;
let _rel = hyper.fs_release().await;
(wr, fr)
};
assert!(
flush_result.is_err(),
"expected flush after write_zero to fail, got wr={:?} flush={:?}",
write_zero_result, flush_result,
);
println!(
"[correctness_write_zero] write_zero={:?} flush={:?} calls={}",
write_zero_result,
flush_result,
interceptor.call_count(),
);
let (persisted_size, body_ok) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let size = hyper.fs_getattr().expect("getattr").st_size as usize;
let mut buf = vec![0u8; original_size];
let _ = hyper.fs_read(0, &mut buf).await.expect("read");
(size, buf == payload)
};
assert_eq!(
persisted_size, original_size,
"write_zero silently committed: expected size {}, got {}",
original_size, persisted_size,
);
assert!(body_ok, "original bytes corrupted after failed write_zero");
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_write_aligned_batch_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let (batch_result, flush_result) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let block0 = AlignedDataBlockWrapper::new(0, 4096, false);
block0.as_mut_slice().fill(0xAA);
let block1 = AlignedDataBlockWrapper::new(1, 4096, true); let block2 = AlignedDataBlockWrapper::new(2, 4096, false);
block2.as_mut_slice().fill(0xCC);
let blocks = vec![block0, block1, block2];
let br = hyper.fs_write_aligned_batch(blocks).await;
let fr = hyper.fs_flush().await;
let _rel = hyper.fs_release().await;
(br, fr)
};
assert!(
flush_result.is_err(),
"expected flush after write_aligned_batch to fail, got batch={:?} flush={:?}",
batch_result, flush_result,
);
println!(
"[correctness_aligned_batch] batch={:?} flush={:?} calls={}",
batch_result,
flush_result,
interceptor.call_count(),
);
let persisted_size = {
let hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
hyper.fs_getattr().expect("getattr").st_size as usize
};
assert_eq!(
persisted_size, 0,
"write_aligned_batch silently committed: expected 0, got {}",
persisted_size,
);
tf.cleanup(&client).await;
}
#[tokio::test]
#[ignore]
async fn rollback_correctness_write_batch_single_flush_fails() {
let _ = env_logger::try_init();
let client = make_client().await;
let tf = TestFile::new(&client).await;
let original_size = 4096usize;
let payload: Vec<u8> = (0..original_size).map(|i| (i & 0xFF) as u8).collect();
{
let mut hyper = Hyper::fs_open_or_create_with_default_opt(
&client,
tf.uri(),
FileFlags::rdwr(),
FileMode::default_file(),
)
.await
.expect("create");
let _ = hyper.fs_write(0, &payload).await.expect("write");
let _ = hyper.fs_release().await.expect("release");
}
let interceptor = FailOnFlushInode::at(1);
let (batch_result, flush_result) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr())
.await
.expect("open");
hyper.with_staging_interceptor(interceptor.clone());
let part = BatchDataBlockWrapper::new_partial_block(
0, 4096, 100, 200, false,
);
part.as_mut_slice().fill(0xFF);
let blocks = vec![part];
let br = hyper.fs_write_batch(blocks).await;
let fr = hyper.fs_flush().await;
let _rel = hyper.fs_release().await;
(br, fr)
};
assert!(
flush_result.is_err(),
"expected flush after write_batch to fail, got batch={:?} flush={:?}",
batch_result, flush_result,
);
println!(
"[correctness_batch] batch={:?} flush={:?} calls={}",
batch_result,
flush_result,
interceptor.call_count(),
);
let (persisted_size, body_ok) = {
let mut hyper = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly())
.await
.expect("reopen");
let size = hyper.fs_getattr().expect("getattr").st_size as usize;
let mut buf = vec![0u8; original_size];
let _ = hyper.fs_read(0, &mut buf).await.expect("read");
(size, buf == payload)
};
assert_eq!(
persisted_size, original_size,
"write_batch silently changed size: expected {}, got {}",
original_size, persisted_size,
);
assert!(
body_ok,
"write_batch silently modified original bytes",
);
tf.cleanup(&client).await;
}