gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 22:37:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): reuse pinned directory handles for object I/O

Browse Source
This commit is contained in:
gatewayJ
2026-09-28 00:55:32 +08:00
committed by Hauser
parent 2aa49aefc7
commit d646f85778
7 changed files with 1172 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/fs.rs
-28
View File
@@ -223,12 +223,6 @@ pub fn rename_std(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()
std::fs::rename(from, to)
}
#[cfg(any(not(windows), test))]
#[tracing::instrument(level = "debug", skip_all)]
pub async fn read_file(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
fs::read(path.as_ref()).await
}
// Bucket existence cache - reduces statx syscalls for repeated bucket checks
/// Cache for bucket directory existence checks.
@@ -614,28 +608,6 @@ mod tests {
assert_eq!(content, "test content");
}
#[tokio::test]
async fn test_read_file() {
let temp_dir = TempDir::new().unwrap();
let file_path = temp_dir.path().join("test_read.txt");
let test_content = b"This is test content for reading";
tokio::fs::write(&file_path, test_content).await.unwrap();
// Read file
let read_content = read_file(&file_path).await.unwrap();
assert_eq!(read_content, test_content);
}
#[tokio::test]
async fn test_read_file_nonexistent() {
let temp_dir = TempDir::new().unwrap();
let file_path = temp_dir.path().join("nonexistent.txt");
// Should fail for non-existent file
assert!(read_file(&file_path).await.is_err());
}
#[tokio::test]
async fn test_same_file() {
let temp_dir = TempDir::new().unwrap();
crates/ecstore/src/disk/local.rs
+99 -91
View File
@@ -189,8 +189,10 @@ struct ListingMetadataRead {
has_namespace_child_candidate: bool,
}
fn read_all_data_std(path: &Path) -> core::result::Result<(Vec<u8>, Option<OffsetDateTime>), ReadAllError> {
let mut file = std::fs::File::open(path).map_err(ReadAllError::Open)?;
fn read_all_data_std(
opened: std::io::Result<std::fs::File>,
) -> core::result::Result<(Vec<u8>, Option<OffsetDateTime>), ReadAllError> {
let mut file = opened.map_err(ReadAllError::Open)?;
let metadata = file.metadata().map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
if metadata.is_dir() {
@@ -5309,24 +5311,6 @@ fn mount_id_from_mountinfo_contents(mountinfo: &str, path: &Path) -> Option<u64>
}
impl LocalDisk {
#[cfg(target_os = "linux")]
fn open_mount_lease(root: &Path) -> Result<(std::fs::File, PathBuf, Option<u64>)> {
use rustix::fs::{Mode, OFlags, open};
use std::os::fd::AsRawFd as _;
let fd = open(
root,
OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
Mode::empty(),
)
.map_err(std::io::Error::from)
.map_err(DiskError::from)?;
let lease = std::fs::File::from(fd);
let io_root = PathBuf::from(format!("/proc/self/fd/{}/.", lease.as_raw_fd()));
let mount_id = mount_id_for_fd(&lease);
Ok((lease, io_root, mount_id))
}
#[cfg(not(target_os = "linux"))]
fn open_mount_lease(root: &Path) -> Result<PathBuf> {
Ok(root.to_path_buf())
@@ -5410,7 +5394,11 @@ impl LocalDisk {
let root = publication_root.path().to_path_buf();
#[cfg(target_os = "linux")]
let (mount_lease, io_root, mount_lease_mount_id) = Self::open_mount_lease(&root)?;
let (mount_lease, io_root, mount_lease_mount_id) = (
publication_root.directory().clone(),
publication_root.io_path().to_path_buf(),
mount_id_for_fd(publication_root.directory()),
);
#[cfg(not(target_os = "linux"))]
let io_root = Self::open_mount_lease(&root)?;
@@ -5458,9 +5446,11 @@ impl LocalDisk {
state = "format_path_resolved",
"Local disk format path resolved"
);
let (format_data, format_meta) = read_file_exists(&io_format_path).await.inspect_err(|err| {
log_startup_disk_error("read_format_json", &io_format_path, err);
})?;
let (format_data, format_meta) = read_file_exists(&io_format_path, &publication_root)
.await
.inspect_err(|err| {
log_startup_disk_error("read_format_json", &io_format_path, err);
})?;
let mut id = None;
// let mut format_legacy = false;
@@ -5551,7 +5541,7 @@ impl LocalDisk {
publication_root,
io_root: io_root.clone(),
#[cfg(target_os = "linux")]
mount_lease: Arc::new(mount_lease),
mount_lease,
#[cfg(target_os = "linux")]
mount_lease_mount_id,
endpoint: ep.clone(),
@@ -6008,6 +5998,19 @@ impl LocalDisk {
self.local_disk_object_path(self.io_root(), bucket, key)
}
/// Only for operations that open through PublicationRoot or a commit guard:
/// containment and symlink checks happen with the actual open, off the runtime.
fn io_get_object_open_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
#[cfg(target_os = "linux")]
{
let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
Ok(path)
}
#[cfg(not(target_os = "linux"))]
self.io_get_object_path(bucket, key)
}
fn io_get_bucket_path(&self, bucket: &str) -> Result<PathBuf> {
self.local_disk_bucket_path(self.io_root(), bucket)
}
@@ -6166,7 +6169,7 @@ impl LocalDisk {
let mut meta = FileMeta::new();
if !fi.fresh {
let (buf, _) = read_file_exists(&p).await?;
let (buf, _) = read_file_exists(&p, &self.publication_root).await?;
if !buf.is_empty() {
let _ = meta.unmarshal_msg(&buf).map_err(|_| {
meta = FileMeta::new();
@@ -7017,9 +7020,10 @@ impl LocalDisk {
// - metadata failure -> to_file_error
// - parse failure -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
let path = file_path.as_ref().to_path_buf();
let root = self.publication_root.clone();
let (data, modtime) = tokio::task::spawn_blocking(move || -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
// Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
let mut f = std::fs::File::open(&path).map_err(to_file_error)?;
let mut f = root.open_readonly(&path).map_err(to_file_error)?;
let meta = f.metadata().map_err(to_file_error)?;
@@ -7072,6 +7076,7 @@ impl LocalDisk {
let metadata_path = self.io_get_object_path(bucket, path.to_string_lossy().as_ref());
// A part's existence check, metadata read and decode share one dispatch.
// Keep open errors unmapped for the existing missing-volume fallback.
let root = self.publication_root.clone();
let result = tokio::task::spawn_blocking(move || -> Result<_> {
let part_error = |error: String| ObjectPartInfo {
number: num,
@@ -7084,7 +7089,7 @@ impl LocalDisk {
// Invalid metadata paths remain request errors, but missing data wins
// first, as it does in the serial reader.
let metadata_path = metadata_path?;
Ok(read_all_data_std(&metadata_path)
Ok(read_all_data_std(root.open_readonly(&metadata_path))
.map(|(data, _)| ObjectPartInfo::unmarshal(&data).unwrap_or_else(|err| part_error(err.to_string()))))
})
.await;
@@ -7103,8 +7108,9 @@ impl LocalDisk {
let object_dir = self.io_get_object_path(volume, object_name)?;
let metadata_path = object_dir.join(STORAGE_FORMAT_FILE);
let volume_dir = self.io_get_bucket_path(volume)?;
let root = self.publication_root.clone();
let result = tokio::task::spawn_blocking(move || {
let (bytes, _) = read_all_data_std(&metadata_path)?;
let (bytes, _) = read_all_data_std(root.open_readonly(&metadata_path))?;
let file_meta = FileMeta::load(&bytes).ok();
let data_dirs: HashSet<String> = file_meta
.as_ref()
@@ -7194,7 +7200,8 @@ impl LocalDisk {
// gating the volume fallback on the open error alone is equivalent to
// the original code, where the fallback lived solely in the open match arm.
let path = file_path.as_ref().to_path_buf();
let res = tokio::task::spawn_blocking(move || read_all_data_std(&path))
let root = self.publication_root.clone();
let res = tokio::task::spawn_blocking(move || read_all_data_std(root.open_readonly(&path)))
.await
.map_err(DiskError::from)?;
@@ -7926,16 +7933,16 @@ impl LocalDisk {
os::make_dir_all(parent, skip_parent).await?;
}
let root = self.publication_root.clone();
tokio::task::spawn_blocking(move || {
#[cfg(test)]
run_owned_file_write_before_open(&path);
let mut file = std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&path)
.map_err(to_file_error)?;
let (mut file, parent_handle) = if sync == SyncMode::FileAndDir {
root.create_truncate_with_parent(&path).map_err(to_file_error)?
} else {
(root.create_truncate(&path).map_err(to_file_error)?, None)
};
std::io::Write::write_all(&mut file, buf.as_ref()).map_err(to_file_error)?;
if sync != SyncMode::None {
file.sync_data().map_err(to_file_error)?;
@@ -7944,7 +7951,7 @@ impl LocalDisk {
if sync == SyncMode::FileAndDir
&& let Some(parent) = path.parent()
{
os::fsync_dir_std(parent).map_err(to_file_error)?;
os::fsync_directory_handle_std(parent, parent_handle.as_ref()).map_err(to_file_error)?;
}
}
Ok::<_, std::io::Error>(())
@@ -7969,7 +7976,18 @@ impl LocalDisk {
os::make_dir_all(parent, skip_parent).await?;
}
let f = super::fs::open_file(path.as_ref(), mode).await.map_err(to_file_error)?;
let f = if mode == O_CREATE | O_WRONLY | O_TRUNC {
let path = path.as_ref().to_path_buf();
let root = self.publication_root.clone();
File::from_std(
tokio::task::spawn_blocking(move || root.create_truncate(&path))
.await
.map_err(DiskError::from)?
.map_err(to_file_error)?,
)
} else {
super::fs::open_file(path.as_ref(), mode).await.map_err(to_file_error)?
};
Ok(f)
}
@@ -8835,9 +8853,9 @@ fn file_meta_counts_toward_limit(meta: &FileMeta) -> bool {
}
// Filter std::io::ErrorKind::NotFound
async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Bytes, Option<Metadata>)> {
async fn read_file_exists(path: impl AsRef<Path>, root: &os::PublicationRoot) -> Result<(Bytes, Option<Metadata>)> {
let p = path.as_ref();
let (data, meta) = match read_file_all(&p).await {
let (data, meta) = match read_file_all(&p, root).await {
Ok((data, meta)) => (data, Some(meta)),
Err(e) => {
if e == Error::FileNotFound {
@@ -8856,32 +8874,28 @@ async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Bytes, Option<Metad
Ok((data, meta))
}
async fn read_file_all(path: impl AsRef<Path>) -> Result<(Bytes, Metadata)> {
let p = path.as_ref();
let meta = read_file_metadata(&path).await?;
let data = fs::read(&p)
.await
.inspect_err(|err| {
log_startup_disk_io_error("read_file_all", p, err);
})
.map_err(to_file_error)?;
Ok((data.into(), meta))
}
async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata> {
let path = p.as_ref();
let meta = fs::metadata(path)
.await
.inspect_err(|err| {
if err.kind() != ErrorKind::NotFound {
log_startup_disk_io_error("read_file_metadata", path, err);
}
})
.map_err(to_file_error)?;
Ok(meta)
async fn read_file_all(path: impl AsRef<Path>, root: &os::PublicationRoot) -> Result<(Bytes, Metadata)> {
let path = path.as_ref().to_path_buf();
let root = root.clone();
tokio::task::spawn_blocking(move || {
let result = (|| {
let mut file = root.open_readonly(&path)?;
let metadata = file.metadata()?;
let mut data = Vec::new();
std::io::Read::read_to_end(&mut file, &mut data)?;
Ok::<_, std::io::Error>((Bytes::from(data), metadata))
})();
result
.inspect_err(|err| {
if err.kind() != ErrorKind::NotFound {
log_startup_disk_io_error("read_file_all", &path, err);
}
})
.map_err(to_file_error)
.map_err(DiskError::from)
})
.await
.map_err(DiskError::from)?
}
fn skip_access_checks(p: impl AsRef<str>) -> bool {
@@ -8963,13 +8977,19 @@ fn check_local_disk_valid_path(root: &Path, path: impl AsRef<Path>) -> Result<()
#[cfg(target_os = "linux")]
fn check_local_disk_valid_object_path_at(root: &Path, root_fd: &std::fs::File, bucket_path: &Path, path: &Path) -> Result<()> {
check_local_disk_object_path_components(root, bucket_path, path)?;
reject_local_disk_symlink_components_at(root, root_fd, &normalize_path_components(path))
}
#[cfg(target_os = "linux")]
fn check_local_disk_object_path_components(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> {
let bucket_path = normalize_path_components(bucket_path);
let path = normalize_path_components(path);
if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
return Err(DiskError::InvalidPath);
}
reject_local_disk_symlink_components_at(root, root_fd, &path)
Ok(())
}
#[cfg(target_os = "linux")]
@@ -9519,9 +9539,9 @@ impl DiskAPI for LocalDisk {
}
}
let p = self.io_get_object_path(volume, path)?;
let p = self.io_get_object_open_path(volume, path)?;
let (data, _) = read_file_all(&p).await?;
let (data, _) = read_file_all(&p, &self.publication_root).await?;
Ok(data)
}
@@ -11120,7 +11140,7 @@ impl DiskAPI for LocalDisk {
};
// if req.metadata_only {}
match read_file_all(&fpath).await {
match read_file_all(&fpath, &self.publication_root).await {
Ok((data, meta)) => {
found += 1;
@@ -21382,10 +21402,12 @@ mod test {
#[tokio::test]
async fn test_read_file_exists() {
let test_file = "./test_read_exists.txt";
let directory = tempfile::tempdir().expect("test directory");
let root = os::PublicationRoot::new(directory.path()).expect("publication root");
let test_file = &directory.path().join("read_exists");
// Test non-existent file
let (data, metadata) = read_file_exists(test_file).await.expect("operation should succeed");
let (data, metadata) = read_file_exists(test_file, &root).await.expect("operation should succeed");
assert!(data.is_empty());
assert!(metadata.is_none());
@@ -21393,7 +21415,7 @@ mod test {
fs::write(test_file, b"test content").await.expect("operation should succeed");
// Test existing file
let (data, metadata) = read_file_exists(test_file).await.expect("operation should succeed");
let (data, metadata) = read_file_exists(test_file, &root).await.expect("operation should succeed");
assert_eq!(data.as_ref(), b"test content");
assert!(metadata.is_some());
@@ -21403,14 +21425,16 @@ mod test {
#[tokio::test]
async fn test_read_file_all() {
let test_file = "./test_read_all.txt";
let directory = tempfile::tempdir().expect("test directory");
let root = os::PublicationRoot::new(directory.path()).expect("publication root");
let test_file = &directory.path().join("read_all");
let test_content = b"test content for read_all";
// Create test file
fs::write(test_file, test_content).await.expect("operation should succeed");
// Test reading file
let (data, metadata) = read_file_all(test_file).await.expect("operation should succeed");
let (data, metadata) = read_file_all(test_file, &root).await.expect("operation should succeed");
assert_eq!(data.as_ref(), test_content);
assert!(metadata.is_file());
assert_eq!(metadata.len(), test_content.len() as u64);
@@ -21419,22 +21443,6 @@ mod test {
let _ = fs::remove_file(test_file).await;
}
#[tokio::test]
async fn test_read_file_metadata() {
let test_file = "./test_metadata.txt";
// Create test file
fs::write(test_file, b"test").await.expect("operation should succeed");
// Test reading metadata
let metadata = read_file_metadata(test_file).await.expect("operation should succeed");
assert!(metadata.is_file());
assert_eq!(metadata.len(), 4); // "test" is 4 bytes
// Clean up
let _ = fs::remove_file(test_file).await;
}
#[test]
fn test_is_root_path() {
// Unix root path
crates/ecstore/src/disk/local/commit.rs
+45 -37
View File
@@ -173,6 +173,7 @@ pub(super) fn create_local_inline_rollback_backup(
Ok(backup_path)
}
#[cfg(test)]
pub(super) async fn lock_rename_commit_directories(
source_parent: &Path,
destination_parent: &Path,
@@ -180,7 +181,7 @@ pub(super) async fn lock_rename_commit_directories(
publication_root: &os::PublicationRoot,
mutation_lease: Arc<os::NamespaceMutationLease>,
) -> Result<os::RenameCommitGuard> {
#[cfg(windows)]
#[cfg(any(windows, target_os = "linux"))]
let result = {
let source_parent = source_parent.to_path_buf();
let destination_parent = destination_parent.to_path_buf();
@@ -188,7 +189,7 @@ pub(super) async fn lock_rename_commit_directories(
let publication_root = publication_root.clone();
os::run_blocking_namespace_operation(mutation_lease, move || {
let result = os::prepare_rename_commit_guard(&source_parent, &destination_parent, &base_dir, &publication_root);
#[cfg(test)]
#[cfg(all(test, windows))]
if result.is_ok() {
run_destination_commit_directory_preparation(&destination_parent);
}
@@ -196,7 +197,7 @@ pub(super) async fn lock_rename_commit_directories(
})
.await
};
#[cfg(not(windows))]
#[cfg(not(any(windows, target_os = "linux")))]
let result = {
let _ = mutation_lease;
os::prepare_rename_commit_guard(source_parent, destination_parent, base_dir, publication_root)
@@ -210,33 +211,42 @@ pub(super) async fn lock_rename_commit_directories(
result.map_err(to_file_error).map_err(DiskError::from)
}
async fn read_rename_destination_metadata(
async fn prepare_rename_destination_metadata(
source_parent: &Path,
file_path: &Path,
rename_commit_guard: &os::RenameCommitGuard,
base_dir: &Path,
publication_root: &os::PublicationRoot,
mutation_lease: Arc<os::NamespaceMutationLease>,
) -> Result<Option<Bytes>> {
#[cfg(windows)]
let result = {
let file_path = file_path.to_path_buf();
let rename_commit_guard = rename_commit_guard.clone();
os::run_blocking_namespace_operation(mutation_lease, move || {
os::read_destination_file_with_commit_guard(&file_path, &rename_commit_guard)
})
.await
};
#[cfg(not(windows))]
let _ = (rename_commit_guard, mutation_lease);
#[cfg(not(windows))]
let result = match super::super::fs::read_file(file_path).await {
Ok(data) => Ok(Some(data)),
Err(err) if err.kind() == ErrorKind::NotFound => Ok(None),
Err(err) => Err(err),
};
result
.map(|data| data.map(Bytes::from))
.map_err(to_file_error)
.map_err(DiskError::from)
) -> Result<(os::RenameCommitGuard, Option<Bytes>)> {
let source_parent = source_parent.to_path_buf();
let file_path = file_path.to_path_buf();
let base_dir = base_dir.to_path_buf();
let publication_root = publication_root.clone();
os::run_blocking_namespace_operation(mutation_lease, move || {
let parent = file_path
.parent()
.ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "missing metadata parent"))?;
let guard = os::prepare_rename_commit_guard(&source_parent, parent, &base_dir, &publication_root).map_err(|err| {
match std::fs::symlink_metadata(&base_dir) {
Err(base_err) if base_err.kind() == ErrorKind::NotFound => base_err,
_ => err,
}
})?;
#[cfg(all(test, windows))]
run_destination_commit_directory_preparation(parent);
#[cfg(any(windows, target_os = "linux"))]
let data = os::read_destination_file_with_commit_guard(&file_path, &guard)?;
#[cfg(not(any(windows, target_os = "linux")))]
let data = match std::fs::read(&file_path) {
Ok(data) => Some(data),
Err(err) if err.kind() == ErrorKind::NotFound => None,
Err(err) => return Err(err),
};
Ok((guard, data.map(Bytes::from)))
})
.await
.map_err(to_file_error)
.map_err(DiskError::from)
}
async fn restore_renamed_data_source(
@@ -419,7 +429,7 @@ impl LocalDisk {
// xl.meta path
let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?;
let dst_file_path = self.io_get_object_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?;
let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?;
// data_dir path
let has_data_dir_path = {
@@ -473,18 +483,16 @@ impl LocalDisk {
if !no_inline {
fs::create_dir_all(src_file_parent).await.map_err(to_file_error)?;
}
// Acquire the common trees before reading destination metadata. On
// Windows this pins the object directory identity across metadata
// preparation, data publication, rollback backup, and final commit.
let rename_commit_guard = lock_rename_commit_directories(
// Open the commit directories and read the old metadata in one blocking
// operation. The returned guard owns the same identities through publish.
let (rename_commit_guard, has_dst_buf) = prepare_rename_destination_metadata(
src_file_parent,
dst_file_parent,
&dst_file_path,
&dst_volume_dir,
&self.publication_root,
mutation_lease.clone(),
)
.await?;
let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?;
if no_inline {
// Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta
@@ -858,7 +866,7 @@ impl LocalDisk {
&& let Some(parent) = dst_file_path.parent()
{
let fsync_started = rustfs_io_metrics::put_stage_timer();
if let Err(err) = os::fsync_dst_dir_group_commit(parent, Some(mutation_lease.clone())).await {
if let Err(err) = os::fsync_commit_directory(parent, &rename_commit_guard, mutation_lease.clone(), None).await {
rustfs_io_metrics::record_put_object_stage_duration_from(
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
fsync_started,
@@ -1222,7 +1230,7 @@ impl LocalDisk {
{
let fsync_started = rustfs_io_metrics::put_stage_timer();
if let Err(err) =
os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission)
os::fsync_commit_directory(dst_parent, &rename_commit_guard, mutation_lease.clone(), Some(admission))
.await
{
rustfs_io_metrics::record_put_object_stage_duration_from(
crates/ecstore/src/disk/os.rs
+754 -68
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/set_disk/ops/object.rs
+25 -3
View File
@@ -3804,6 +3804,10 @@ impl SetDisks {
);
}
fn needs_old_data_cleanup_receipt(old_data_dir: Uuid, committed_data_dir: Option<Uuid>, epoch: Option<Uuid>) -> bool {
epoch.is_some() && committed_data_dir != Some(old_data_dir)
}
pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts(
&self,
disks: &[Option<DiskStore>],
@@ -3813,10 +3817,10 @@ impl SetDisks {
committed_data_dir: Option<Uuid>,
epoch: Option<Uuid>,
) {
let Some(epoch) = epoch else { return };
if committed_data_dir == Some(old_data_dir) {
if !Self::needs_old_data_cleanup_receipt(old_data_dir, committed_data_dir, epoch) {
return;
}
let Some(epoch) = epoch else { return };
let receipt = OldDataCleanupReceipt::new(epoch, old_data_dir, committed_data_dir);
let Ok(encoded) = receipt.encode() else {
return;
@@ -5540,7 +5544,9 @@ impl SetDisks {
let rename_stage_elapsed = rename_stage_start.elapsed();
let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
if let Some(old_dir) = op_old_dir {
if let Some(old_dir) = op_old_dir
&& Self::needs_old_data_cleanup_receipt(old_dir, committed_data_dir, transaction_epoch)
{
rustfs_io_metrics::record_put_object_stage_duration(
rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_CLEANUP_RECEIPT_TRIGGERED,
1.0,
@@ -23696,3 +23702,19 @@ mod single_delete_namespace_owner_tests {
assert_single_delete_physical_owner("rollback").await;
}
}
#[cfg(test)]
mod cleanup_receipt_admission_tests {
use super::*;
#[test]
fn receipt_admission_skips_only_work_that_the_writer_would_skip() {
let old = Uuid::new_v4();
let current = Uuid::new_v4();
let epoch = Uuid::new_v4();
assert!(!SetDisks::needs_old_data_cleanup_receipt(old, Some(current), None));
assert!(!SetDisks::needs_old_data_cleanup_receipt(old, Some(old), Some(epoch)));
assert!(SetDisks::needs_old_data_cleanup_receipt(old, Some(current), Some(epoch)));
assert!(SetDisks::needs_old_data_cleanup_receipt(old, None, Some(epoch)));
}
}
crates/ecstore/tests/fd_relative_io_test.rs
+243
View File
@@ -0,0 +1,243 @@
// Copyright 2026 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#![cfg(all(target_os = "linux", feature = "test-util"))]
mod storage_api;
use bytes::Bytes;
use rustfs_filemeta::{FileInfo, ObjectPartInfo};
use std::{path::Path, time::Duration};
use storage_api::fd_relative::{
DiskAPI, DiskError, DiskOption, DiskStore, Endpoint, LocalPublicationPause, LocalPublicationStage, ReadOptions, new_disk,
};
use time::OffsetDateTime;
use uuid::Uuid;
const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp";
async fn disk_at(path: &Path) -> DiskStore {
new_disk(
&Endpoint::try_from(path.to_str().expect("UTF-8 test path")).expect("endpoint"),
&DiskOption::default(),
)
.await
.expect("local disk")
}
fn metadata(data: Bytes, inline: bool, directory: Uuid) -> FileInfo {
let mut fi = FileInfo::new("object", 1, 0);
fi.erasure.index = 1;
fi.version_id = Some(Uuid::nil());
fi.data_dir = Some(directory);
fi.size = i64::try_from(data.len()).expect("test data size");
fi.parts = vec![ObjectPartInfo {
number: 1,
size: data.len(),
actual_size: fi.size,
..Default::default()
}];
fi.mod_time = Some(OffsetDateTime::now_utc());
fi.data = inline.then_some(data);
if inline {
fi.set_inline_data();
}
fi
}
async fn put(disk: &DiskStore, staging: &str, data: Bytes, inline: bool) -> Uuid {
let directory = Uuid::new_v4();
if !inline {
disk.write_all(RUSTFS_META_TMP_BUCKET, &format!("{staging}/{directory}/part.1"), data.clone())
.await
.expect("stage shard");
}
disk.rename_data(RUSTFS_META_TMP_BUCKET, staging, metadata(data, inline, directory), "bucket", "object")
.await
.expect("commit object");
directory
}
#[tokio::test]
async fn pinned_root_reads_and_writes_original_disk_after_path_replacement() {
let temp = tempfile::tempdir().expect("test directory");
let root = temp.path().join("disk");
std::fs::create_dir(&root).expect("disk root");
let disk = disk_at(&root).await;
disk.make_volume("bucket").await.expect("bucket");
disk.write_all("bucket", "metadata", Bytes::from_static(b"original"))
.await
.expect("initial write");
let moved = temp.path().join("original");
std::fs::rename(&root, &moved).expect("move mount path");
std::fs::create_dir_all(root.join("bucket")).expect("replacement disk");
std::fs::write(root.join("bucket/metadata"), b"replacement").expect("replacement sentinel");
assert_eq!(
disk.read_all("bucket", "metadata").await.expect("pinned read"),
Bytes::from_static(b"original")
);
disk.write_all("bucket", "metadata", Bytes::from_static(b"updated"))
.await
.expect("pinned write");
assert_eq!(std::fs::read(moved.join("bucket/metadata")).expect("original disk"), b"updated");
assert_eq!(std::fs::read(root.join("bucket/metadata")).expect("replacement disk"), b"replacement");
}
#[tokio::test]
async fn relative_reads_and_writes_reject_symlink_escape() {
let temp = tempfile::tempdir().expect("test directory");
let disk = disk_at(temp.path()).await;
disk.make_volume("bucket").await.expect("bucket");
let outside = tempfile::tempdir().expect("outside directory");
std::fs::write(outside.path().join("sentinel"), b"preserved").expect("sentinel");
std::os::unix::fs::symlink(outside.path(), temp.path().join("bucket/link")).expect("directory link");
std::os::unix::fs::symlink(outside.path().join("sentinel"), temp.path().join("bucket/file-link")).expect("file link");
for path in ["link/sentinel", "file-link"] {
assert!(matches!(disk.read_all("bucket", path).await, Err(DiskError::InvalidPath)));
assert!(matches!(
disk.write_all("bucket", path, Bytes::from_static(b"unsafe")).await,
Err(DiskError::InvalidPath)
));
}
assert_eq!(std::fs::read(outside.path().join("sentinel")).expect("outside bytes"), b"preserved");
}
#[tokio::test]
async fn inline_and_non_inline_overwrites_remain_readable_after_reopen() {
for inline in [true, false] {
let temp = tempfile::tempdir().expect("test directory");
let disk = disk_at(temp.path()).await;
disk.make_volume("bucket").await.expect("bucket");
put(&disk, "first", Bytes::from_static(b"old bytes"), inline).await;
let directory = put(&disk, "second", Bytes::from_static(b"new bytes"), inline).await;
drop(disk);
let disk = disk_at(temp.path()).await;
let fi = disk
.read_version(
"bucket",
"bucket",
"object",
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("committed metadata after reopen");
assert_eq!(fi.data_dir, Some(directory));
if inline {
assert_eq!(fi.data, Some(Bytes::from_static(b"new bytes")));
} else {
assert_eq!(
disk.read_all("bucket", &format!("object/{directory}/part.1"))
.await
.expect("shard bytes"),
Bytes::from_static(b"new bytes")
);
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn prepared_publication_rejects_source_or_directory_replacement() {
for replace_directory in [false, true] {
let temp = tempfile::tempdir().expect("test directory");
let disk = disk_at(temp.path()).await;
disk.make_volume("bucket").await.expect("bucket");
put(&disk, "first", Bytes::from_static(b"old bytes"), true).await;
let old = std::fs::read(temp.path().join("bucket/object/xl.meta")).expect("old metadata");
let mut pause = LocalPublicationPause::install(&disk, "bucket", "object/xl.meta", LocalPublicationStage::PreparedRename)
.expect("publication pause");
let writer = disk.clone();
let task = tokio::spawn(async move {
writer
.rename_data(
RUSTFS_META_TMP_BUCKET,
"second",
metadata(Bytes::from_static(b"new bytes"), true, Uuid::new_v4()),
"bucket",
"object",
)
.await
});
tokio::time::timeout(Duration::from_secs(30), pause.entered())
.await
.expect("publication reached pause")
.expect("pause entered");
if replace_directory {
std::fs::rename(temp.path().join("bucket/object"), temp.path().join("detached")).expect("detach object");
std::fs::create_dir(temp.path().join("bucket/object")).expect("replacement directory");
std::fs::write(temp.path().join("bucket/object/xl.meta"), &old).expect("replacement metadata");
} else {
let src = temp.path().join(RUSTFS_META_TMP_BUCKET).join("second/xl.meta");
std::fs::rename(&src, src.with_extension("detached")).expect("detach staged metadata");
std::fs::write(src, b"unrelated").expect("replacement staged file");
}
drop(pause);
assert!(task.await.expect("writer task").is_err());
assert_eq!(
std::fs::read(temp.path().join("bucket/object/xl.meta")).expect("old metadata survives"),
old
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cancelled_waiter_keeps_publication_serialized_until_physical_completion() {
let temp = tempfile::tempdir().expect("test directory");
let disk = disk_at(temp.path()).await;
disk.make_volume("bucket").await.expect("bucket");
put(&disk, "first", Bytes::from_static(b"old bytes"), true).await;
let mut pause = LocalPublicationPause::install(&disk, "bucket", "object/xl.meta", LocalPublicationStage::PreparedRename)
.expect("publication pause");
let writer = disk.clone();
let pending = tokio::spawn(async move { put(&writer, "shared-stage", Bytes::from_static(b"cancelled waiter"), true).await });
tokio::time::timeout(Duration::from_secs(30), pause.entered())
.await
.expect("publication reached pause")
.expect("pause entered");
pending.abort();
assert!(pending.await.expect_err("waiter aborted").is_cancelled());
let writer = disk.clone();
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let mut retry = tokio::spawn(async move {
let _ = started_tx.send(());
put(&writer, "shared-stage", Bytes::from_static(b"latest bytes"), true).await
});
started_rx.await.expect("retry started");
assert!(
tokio::time::timeout(Duration::from_millis(200), &mut retry).await.is_err(),
"a retry must not publish while the cancelled writer still owns physical IO"
);
drop(pause);
tokio::time::timeout(Duration::from_secs(30), retry)
.await
.expect("retry completed")
.expect("retry task");
let fi = disk
.read_version(
"bucket",
"bucket",
"object",
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("latest metadata");
assert_eq!(fi.data, Some(Bytes::from_static(b"latest bytes")));
}
crates/ecstore/tests/storage_api.rs
+6
View File
@@ -67,3 +67,9 @@ pub(crate) mod cleanup_isolation {
pub(crate) use super::{DiskAPI, DiskOption, Endpoint, new_disk};
pub(crate) use rustfs_ecstore::api::disk::{DeleteOptions, DiskError};
}
#[cfg(all(feature = "test-util", target_os = "linux"))]
pub(crate) mod fd_relative {
pub(crate) use super::{DiskAPI, DiskOption, DiskStore, Endpoint, new_disk};
pub(crate) use rustfs_ecstore::api::disk::{DiskError, LocalPublicationPause, LocalPublicationStage, ReadOptions};
}