gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): move speculative PUT-tail tmp cleanup off the hot path (#4389)

Browse Source 
* perf(ecstore): move speculative tmp cleanup off the PUT hot path

On a successful PUT, rename_data has already moved the data dir out of the tmp workspace, so the delete_all(RUSTFS_META_TMP_BUCKET) at the end of SetDisks::put_object is a speculative no-op safety net. It was awaited inline on the response path, where profiling (backlog#924 / HP-3) showed the same-disk queueing behind fsync-heavy load turns a ~49us no-op into ~9ms average (p99 77ms, macOS F_FULLFSYNC amplified) added to every PUT.

Run that cleanup on a spawned task instead, keeping it as a real backstop (rename_data's remove_std only removes empty dirs and silently ignores failures). The failure path (quorum loss / rollback) keeps the cleanup inline so a failed PUT never returns with tmp shards still on disk. If the process dies before the spawned task runs, cleanup_stale_tmp_objects (24h expiry, 5-minute loop) reclaims the entry.

Scope note: ops/multipart.rs delete_all on RUSTFS_META_MULTIPART_BUCKET is intentionally untouched; it removes real leftovers and deferring it would widen CompleteMultipartUpload/Abort races.

Regression tests (hermetic SetDisks on formatted local disks, no global state): PUT success drains the tmp workspace (polling the spawned task), and PUT failure (missing bucket volume, rename_data quorum error after tmp shards were written) cleans the workspace inline before returning.

Ref: https://github.com/rustfs/backlog/issues/924

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): do not retry NotFound in reliable_rename_inner

reliable_rename_inner blindly retried the rename once on any error. A NotFound retry cannot succeed: nothing recreates the missing source or parent directory between attempts, so the second rename fails identically and speculative cleanup renames (e.g. move_to_trash on an already-removed tmp path) always paid for two syscalls.

Extract the retry decision into should_retry_rename: NotFound returns immediately, any other error keeps the existing single retry. This helper is shared by the rename_data commit path via rename_all, so behavior there is covered by a new rename_all success regression test alongside the retry-predicate tests.

Ref: https://github.com/rustfs/backlog/issues/924

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-08 04:01:37 +08:00
committed by GitHub
parent 062a68d151
commit e7cc719c17
2 changed files with 247 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/os.rs
+46 -1
View File
@@ -241,7 +241,7 @@ async fn reliable_rename_inner(
let mut i = 0;
loop {
if let Err(e) = super::fs::rename_std(src_file_path.as_ref(), dst_file_path.as_ref()) {
if i == 0 {
if should_retry_rename(&e, i) {
i += 1;
continue;
}
@@ -263,6 +263,19 @@ async fn reliable_rename_inner(
Ok(())
}
/// Whether a failed `rename` in [`reliable_rename_inner`] should be retried.
///
/// Only the first failure is retried, and `NotFound` is never retried: the
/// retry does not recreate the missing source or parent directory, so a second
/// attempt is guaranteed to fail identically. Skipping it spares speculative
/// cleanup renames (e.g. `move_to_trash` on an already-removed tmp path) a
/// pointless second syscall. This predicate is shared by the `rename_data`
/// commit path via `rename_all`, so any relaxation here must keep genuine
/// transient errors retryable.
fn should_retry_rename(err: &io::Error, attempt: usize) -> bool {
attempt == 0 && err.kind() != io::ErrorKind::NotFound
}
pub async fn reliable_mkdir_all(path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> io::Result<()> {
let mut i = 0;
@@ -366,6 +379,38 @@ mod tests {
assert!(!dst.exists());
}
#[test]
fn rename_retry_never_retries_not_found() {
// NotFound is terminal for the retry loop: the retry does not recreate
// the missing source/parent, so a second rename would fail identically.
let not_found = io::Error::new(io::ErrorKind::NotFound, "missing");
assert!(!should_retry_rename(&not_found, 0));
assert!(!should_retry_rename(&not_found, 1));
}
#[test]
fn rename_retry_allows_single_retry_for_other_errors() {
let denied = io::Error::new(io::ErrorKind::PermissionDenied, "denied");
assert!(should_retry_rename(&denied, 0));
assert!(!should_retry_rename(&denied, 1));
}
#[tokio::test]
async fn rename_all_moves_existing_directory_tree() {
// Guards the rename_data commit path, which funnels through
// reliable_rename_inner via rename_all.
let temp_dir = tempdir().expect("create temp dir");
let src = temp_dir.path().join("src-dir");
std::fs::create_dir_all(src.join("nested")).expect("create src tree");
std::fs::write(src.join("nested").join("part.1"), b"payload").expect("write part");
let dst = temp_dir.path().join("dst-parent").join("dst-dir");
rename_all(&src, &dst, temp_dir.path()).await.expect("rename must succeed");
assert!(!src.exists());
assert_eq!(std::fs::read(dst.join("nested").join("part.1")).expect("read moved part"), b"payload");
}
#[tokio::test]
async fn fsync_dir_succeeds_on_directory() {
let temp_dir = tempdir().expect("create temp dir");
crates/ecstore/src/set_disk/ops/object.rs
+201 -10
View File
@@ -1029,16 +1029,45 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
);
}
if let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await {
warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
if result.is_ok() {
// Success path: `rename_data` has already moved the data dir out of
// the tmp workspace and removed the (empty) tmp dir where it could,
// so this delete_all is a speculative safety net that normally hits
// a missing path. It still must run — `rename_data`'s `remove_std`
// only removes empty directories and silently ignores failures —
// but it has no reason to run on the response path: under
// fsync-heavy load the same-disk queueing behind it was measured to
// add ~9ms average (p99 77ms) to PUT latency (backlog#924 / HP-3).
// If the process dies before the spawned task runs, the stale tmp
// entry is reclaimed by cleanup_stale_tmp_objects (24h expiry,
// 5-minute background loop).
let set_disks = self.clone();
tokio::spawn(async move {
if let Err(err) = set_disks.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await {
warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
tmp_dir = %tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
});
} else {
// Failure path (quorum loss / rollback): keep the cleanup inline so
// a failed PUT never returns while its tmp shards are still on disk
// (state-residue hardening tracked by backlog#864 / backlog#898).
if let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await {
warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
}
result
@@ -2253,3 +2282,165 @@ mod b3_write_quorum_tests {
assert_eq!(disks, vec![Some(0), Some(1), Some(2)]);
}
}
#[cfg(test)]
mod put_object_tmp_cleanup_tests {
//! Regression coverage for backlog#924 (HP-3): the speculative tmp-dir
//! cleanup at the end of a successful PUT runs on a spawned task (off the
//! response path), while a failed PUT must still clean its tmp shards
//! inline before returning.
//!
//! The `SetDisks` under test is constructed directly on formatted local
//! disks (same pattern as the `ops/locking.rs` tests) so the tests stay
//! hermetic: no global local-disk registry, lock clients, or ECStore
//! instance is touched, and each test owns its tmp workspace.
use super::*;
use crate::disk::DiskAPI as _;
use crate::store::init_format::save_format_file;
use std::time::Duration;
use tempfile::TempDir;
use tokio::sync::RwLock;
/// Large enough that the erasure shards are written as real tmp files
/// (never inlined into xl.meta), so both tests exercise actual cleanup.
const TEST_OBJECT_SIZE: usize = 1 << 20;
async fn make_formatted_local_disk(disk_idx: usize, format: &FormatV3) -> (TempDir, Endpoint, DiskStore) {
let dir = tempfile::tempdir().expect("tempdir should be created");
let mut endpoint =
Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_idx);
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("local disk should be created");
let mut disk_format = format.clone();
disk_format.erasure.this = format.erasure.sets[0][disk_idx];
save_format_file(&Some(disk.clone()), &Some(disk_format))
.await
.expect("format should be saved");
(dir, endpoint, disk)
}
async fn hermetic_set_disks(disk_count: usize) -> (Vec<TempDir>, Vec<DiskStore>, Arc<SetDisks>) {
let format = FormatV3::new(1, disk_count);
let mut temp_dirs = Vec::with_capacity(disk_count);
let mut endpoints = Vec::with_capacity(disk_count);
let mut disk_stores = Vec::with_capacity(disk_count);
let mut disks = Vec::with_capacity(disk_count);
for disk_idx in 0..disk_count {
let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
temp_dirs.push(temp_dir);
endpoints.push(endpoint);
disk_stores.push(disk.clone());
disks.push(Some(disk));
}
let set_disks = SetDisks::new(
"tmp-cleanup-test-owner".to_string(),
Arc::new(RwLock::new(disks)),
disk_count,
disk_count / 2,
0,
0,
endpoints,
format,
Vec::new(),
)
.await;
(temp_dirs, disk_stores, set_disks)
}
/// Entries under `.rustfs.sys/tmp` on every disk, excluding the `.trash`
/// staging directory (trash reclamation is a background concern).
async fn non_trash_tmp_entries(temp_dirs: &[TempDir]) -> Vec<String> {
let mut leftovers = Vec::new();
for temp_dir in temp_dirs {
let tmp_path = temp_dir.path().join(RUSTFS_META_TMP_BUCKET);
let mut read_dir = match tokio::fs::read_dir(&tmp_path).await {
Ok(read_dir) => read_dir,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
Err(err) => panic!("tmp dir {tmp_path:?} should be listable: {err}"),
};
while let Some(entry) = read_dir.next_entry().await.expect("tmp dir entry should be readable") {
let name = entry.file_name().to_string_lossy().to_string();
if name != ".trash" {
leftovers.push(format!("{}/{name}", tmp_path.display()));
}
}
}
leftovers
}
#[tokio::test]
async fn put_object_success_eventually_cleans_tmp_workspace() {
let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "tmp-clean-ok-bucket";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let mut reader = PutObjReader::from_vec(vec![7u8; TEST_OBJECT_SIZE]);
set_disks
.put_object(bucket, "hot-path-object", &mut reader, &ObjectOptions::default())
.await
.expect("put_object should succeed");
// The speculative cleanup runs on a spawned task off the PUT response
// path, so poll for the tmp workspace to drain instead of asserting
// immediately.
let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
loop {
let leftovers = non_trash_tmp_entries(&temp_dirs).await;
if leftovers.is_empty() {
break;
}
assert!(
tokio::time::Instant::now() < deadline,
"tmp workspace should drain after a successful PUT, leftovers: {leftovers:?}"
);
tokio::time::sleep(Duration::from_millis(25)).await;
}
drop(temp_dirs);
}
#[tokio::test]
async fn put_object_failure_cleans_tmp_workspace_inline() {
let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
// The bucket volume is never created, so the shards are written into
// the tmp workspace and the commit fails at rename_data with a quorum
// error — exercising the failure-path cleanup.
let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
let err = set_disks
.put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default())
.await
.expect_err("put_object into a missing bucket volume must fail");
// No polling: the failure path must clean the tmp workspace inline,
// before put_object returns (backlog#864 / backlog#898 hardening).
let leftovers = non_trash_tmp_entries(&temp_dirs).await;
assert!(
leftovers.is_empty(),
"failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
);
drop(temp_dirs);
}
}