mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
feat(ecstore): reclaim orphaned data dirs during heal (#4356)
* feat(ecstore): reclaim orphaned data dirs during heal (backlog #3231/#3191) Pre-#3510, an unversioned overwrite leaked one UUID-named data dir per PUT. The write path now cleans up going forward, but pre-existing strays stay on disk forever: heal's dangling logic only removes whole objects whose data is missing, never surplus data dirs of an otherwise-healthy object. That leaked space is what eventually made ListObjects scan tens of GB and time out. Add SetDisks::reclaim_orphan_data_dirs, a fail-closed, quorum-safe sweep that mirrors purge_orphan_dir_object: - referenced set is the UNION of get_data_dirs() across every online replica's xl.meta, so a dir named by any replica is kept; - if a disk holds the object dir but its xl.meta is missing or unparseable the object is treated as degraded and nothing is removed; - only UUID-named subdirectories are ever considered. Wire it into heal_object's healthy tail (under the object write lock) as a best-effort step so the background heal scanner and admin heal recover the leaked space automatically; a reclaim failure never fails the heal. Covered by four unit tests: removal of an unreferenced dir, no-op when all dirs are referenced, fail-closed abort on missing metadata, and cross-replica union preservation. Co-Authored-By: heihutu <heihutu@gmail.com> * docs(ecstore): fix typo unparseable -> unparsable Satisfies the repository typos CI check. Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -54,7 +54,7 @@ use crate::set_disk::shard_source::ShardReadCost;
|
||||
use futures::stream::{FuturesUnordered, StreamExt};
|
||||
use metrics::counter;
|
||||
use std::{
|
||||
collections::{HashMap, VecDeque},
|
||||
collections::{HashMap, HashSet, VecDeque},
|
||||
future::Future,
|
||||
pin::Pin,
|
||||
sync::OnceLock,
|
||||
@@ -3189,6 +3189,163 @@ impl SetDisks {
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Reclaim orphaned physical data directories under `bucket/object` that no
|
||||
/// live version in the object's `xl.meta` references any longer.
|
||||
///
|
||||
/// Before #3510, an unversioned overwrite leaked one UUID-named data dir per
|
||||
/// PUT. The write path now cleans up going forward, but pre-existing strays
|
||||
/// stay on disk forever: `heal`'s dangling logic only removes *whole* objects
|
||||
/// whose data is missing, never surplus data dirs of an otherwise-healthy
|
||||
/// object. This closes that gap so the scanner/heal sweep can recover the
|
||||
/// leaked space automatically (issues #3231, #3191).
|
||||
///
|
||||
/// Safety — fail closed:
|
||||
/// * The set of referenced data dirs is the UNION of `get_data_dirs()` across
|
||||
/// every online disk's `xl.meta`, so a dir named by *any* replica is kept.
|
||||
/// * If a disk holds the object directory but its `xl.meta` is missing or
|
||||
/// unparsable, the object is treated as degraded and NOTHING is removed —
|
||||
/// the unreadable copy could be the only one naming a live data dir, and a
|
||||
/// heal must run first.
|
||||
/// * Only subdirectories whose names parse as a UUID are ever considered;
|
||||
/// removal is non-recursive-safe via a recursive delete of the full stray
|
||||
/// data-dir path only.
|
||||
///
|
||||
/// Returns the number of stray data directories removed across the set. This
|
||||
/// is best-effort maintenance: individual delete failures are logged and
|
||||
/// skipped rather than propagated.
|
||||
pub(crate) async fn reclaim_orphan_data_dirs(&self, bucket: &str, object: &str) -> disk::error::Result<usize> {
|
||||
let disks = self.get_disks_internal().await;
|
||||
|
||||
// Phase 1 (read-only): build the referenced-data-dir union and record the
|
||||
// physical UUID subdirectories present on each disk. Abort on any degraded
|
||||
// copy so a healable object is never stripped of a referenced data dir.
|
||||
let mut referenced: HashSet<Uuid> = HashSet::new();
|
||||
let mut per_disk_dirs: Vec<(usize, Vec<Uuid>)> = Vec::new();
|
||||
let mut healthy_metas = 0usize;
|
||||
|
||||
for (i, disk) in disks.iter().enumerate() {
|
||||
let Some(disk) = disk else { continue };
|
||||
|
||||
let entries = match disk.list_dir("", bucket, object, 0).await {
|
||||
Ok(entries) => entries,
|
||||
// No object directory on this disk: nothing to reclaim here.
|
||||
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => continue,
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
|
||||
// Collect the UUID-named subdirectories: these are the physical data
|
||||
// dirs. Non-directory entries (xl.meta) and non-UUID names are ignored.
|
||||
let mut physical = Vec::new();
|
||||
for entry in &entries {
|
||||
let Some(name) = entry.strip_suffix(SLASH_SEPARATOR) else { continue };
|
||||
if let Ok(uuid) = Uuid::parse_str(name)
|
||||
&& !uuid.is_nil()
|
||||
{
|
||||
physical.push(uuid);
|
||||
}
|
||||
}
|
||||
|
||||
// Read and parse this replica's metadata. A directory that carries data
|
||||
// dirs but no readable xl.meta is degraded — fail closed.
|
||||
let meta_path = path_join_buf(&[object, STORAGE_FORMAT_FILE]);
|
||||
let buf = match disk.read_metadata(bucket, &meta_path).await {
|
||||
Ok(buf) => buf,
|
||||
Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {
|
||||
if physical.is_empty() {
|
||||
// Bare directory with no data dirs and no metadata: leave it
|
||||
// to the orphan-dir / dangling-object heal paths.
|
||||
continue;
|
||||
}
|
||||
warn!(
|
||||
target: "rustfs_ecstore::set_disk",
|
||||
bucket, object,
|
||||
"reclaim_orphan_data_dirs: aborting, data dirs present without xl.meta on a disk"
|
||||
);
|
||||
return Ok(0);
|
||||
}
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
|
||||
let meta = match FileMeta::load(&buf) {
|
||||
Ok(meta) => meta,
|
||||
Err(err) => {
|
||||
warn!(
|
||||
target: "rustfs_ecstore::set_disk",
|
||||
bucket, object, error = %err,
|
||||
"reclaim_orphan_data_dirs: aborting, unparsable xl.meta on a disk"
|
||||
);
|
||||
return Ok(0);
|
||||
}
|
||||
};
|
||||
|
||||
match meta.get_data_dirs() {
|
||||
Ok(dirs) => referenced.extend(dirs.into_iter().flatten().filter(|d| !d.is_nil())),
|
||||
Err(err) => {
|
||||
warn!(
|
||||
target: "rustfs_ecstore::set_disk",
|
||||
bucket, object, error = %err,
|
||||
"reclaim_orphan_data_dirs: aborting, could not decode data dirs from xl.meta"
|
||||
);
|
||||
return Ok(0);
|
||||
}
|
||||
}
|
||||
|
||||
healthy_metas += 1;
|
||||
if !physical.is_empty() {
|
||||
per_disk_dirs.push((i, physical));
|
||||
}
|
||||
}
|
||||
|
||||
// No healthy metadata anywhere: this is not a live object, so surplus dirs
|
||||
// (if any) belong to the dangling-object heal path, not here.
|
||||
if healthy_metas == 0 {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
// Phase 2: delete every physical data dir not referenced by the union.
|
||||
let mut removed = 0usize;
|
||||
for (i, physical) in per_disk_dirs {
|
||||
let Some(disk) = disks[i].as_ref() else { continue };
|
||||
for dir in physical {
|
||||
if referenced.contains(&dir) {
|
||||
continue;
|
||||
}
|
||||
let stray = format!("{object}/{dir}");
|
||||
match disk
|
||||
.delete(
|
||||
bucket,
|
||||
&stray,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(()) => {
|
||||
removed += 1;
|
||||
info!(
|
||||
target: "rustfs_ecstore::set_disk",
|
||||
bucket, object, data_dir = %dir,
|
||||
"reclaim_orphan_data_dirs: removed orphaned data directory"
|
||||
);
|
||||
}
|
||||
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
|
||||
Err(err) => {
|
||||
warn!(
|
||||
target: "rustfs_ecstore::set_disk",
|
||||
bucket, object, data_dir = %dir, error = %err,
|
||||
"reclaim_orphan_data_dirs: failed to remove orphaned data directory"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
pub(in crate::set_disk) async fn check_write_precondition(
|
||||
&self,
|
||||
bucket: &str,
|
||||
|
||||
@@ -4487,6 +4487,151 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
// Build an `xl.meta` under `object_dir` whose versions reference `data_dirs`
|
||||
// (one Object version per data dir, each with its own version id).
|
||||
async fn write_object_meta_with_data_dirs(object_dir: &std::path::Path, bucket: &str, object: &str, data_dirs: &[Uuid]) {
|
||||
fs::create_dir_all(object_dir).await.expect("object dir should be created");
|
||||
let mut meta = FileMeta::default();
|
||||
for data_dir in data_dirs {
|
||||
let mut fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 1);
|
||||
fi.name = object.to_string();
|
||||
fi.version_id = Some(Uuid::new_v4());
|
||||
fi.data_dir = Some(*data_dir);
|
||||
fi.size = 1;
|
||||
fi.mod_time = Some(OffsetDateTime::now_utc());
|
||||
meta.add_version(fi).expect("metadata should accept file info");
|
||||
}
|
||||
let buf = meta.marshal_msg().expect("metadata should encode");
|
||||
fs::write(object_dir.join(STORAGE_FORMAT_FILE), buf)
|
||||
.await
|
||||
.expect("metadata should be written");
|
||||
}
|
||||
|
||||
// #3231/#3191: a data dir on disk that no version references (a pre-#3510
|
||||
// unversioned-overwrite leak) must be reclaimed, leaving the live one intact.
|
||||
#[tokio::test]
|
||||
async fn reclaim_orphan_data_dirs_removes_unreferenced_dir() {
|
||||
let (dir, disk) = make_single_local_disk().await;
|
||||
let root = dir.path();
|
||||
let live = Uuid::new_v4();
|
||||
let orphan = Uuid::new_v4();
|
||||
|
||||
let object_dir = root.join("bucket").join("obj");
|
||||
write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
|
||||
fs::create_dir_all(object_dir.join(live.to_string()))
|
||||
.await
|
||||
.expect("live data dir should be created");
|
||||
fs::write(object_dir.join(live.to_string()).join("part.1"), b"live")
|
||||
.await
|
||||
.expect("live part should be written");
|
||||
fs::create_dir_all(object_dir.join(orphan.to_string()))
|
||||
.await
|
||||
.expect("orphan data dir should be created");
|
||||
fs::write(object_dir.join(orphan.to_string()).join("part.1"), b"stale")
|
||||
.await
|
||||
.expect("orphan part should be written");
|
||||
|
||||
let set = make_set_disks_with(vec![Some(disk)]).await;
|
||||
let removed = set
|
||||
.reclaim_orphan_data_dirs("bucket", "obj")
|
||||
.await
|
||||
.expect("reclaim should succeed");
|
||||
|
||||
assert_eq!(removed, 1, "exactly the unreferenced data dir should be removed");
|
||||
assert!(object_dir.join(live.to_string()).exists(), "referenced data dir must be preserved");
|
||||
assert!(!object_dir.join(orphan.to_string()).exists(), "orphaned data dir must be removed");
|
||||
assert!(object_dir.join(STORAGE_FORMAT_FILE).exists(), "metadata must be preserved");
|
||||
}
|
||||
|
||||
// Nothing to reclaim when every physical data dir is still referenced.
|
||||
#[tokio::test]
|
||||
async fn reclaim_orphan_data_dirs_keeps_referenced_dir() {
|
||||
let (dir, disk) = make_single_local_disk().await;
|
||||
let root = dir.path();
|
||||
let live = Uuid::new_v4();
|
||||
|
||||
let object_dir = root.join("bucket").join("obj");
|
||||
write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
|
||||
fs::create_dir_all(object_dir.join(live.to_string()))
|
||||
.await
|
||||
.expect("live data dir should be created");
|
||||
|
||||
let set = make_set_disks_with(vec![Some(disk)]).await;
|
||||
let removed = set
|
||||
.reclaim_orphan_data_dirs("bucket", "obj")
|
||||
.await
|
||||
.expect("reclaim should succeed");
|
||||
|
||||
assert_eq!(removed, 0, "no data dir should be removed");
|
||||
assert!(object_dir.join(live.to_string()).exists(), "referenced data dir must be preserved");
|
||||
}
|
||||
|
||||
// Fail closed: a data dir present without a readable xl.meta is degraded, and
|
||||
// must never be removed (a heal has to run first).
|
||||
#[tokio::test]
|
||||
async fn reclaim_orphan_data_dirs_aborts_when_meta_missing() {
|
||||
let (dir, disk) = make_single_local_disk().await;
|
||||
let root = dir.path();
|
||||
let stray = Uuid::new_v4();
|
||||
|
||||
let object_dir = root.join("bucket").join("obj");
|
||||
fs::create_dir_all(object_dir.join(stray.to_string()))
|
||||
.await
|
||||
.expect("data dir should be created");
|
||||
fs::write(object_dir.join(stray.to_string()).join("part.1"), b"data")
|
||||
.await
|
||||
.expect("part should be written");
|
||||
|
||||
let set = make_set_disks_with(vec![Some(disk)]).await;
|
||||
let removed = set
|
||||
.reclaim_orphan_data_dirs("bucket", "obj")
|
||||
.await
|
||||
.expect("reclaim should succeed");
|
||||
|
||||
assert_eq!(removed, 0, "must not remove data dirs when metadata is absent");
|
||||
assert!(
|
||||
object_dir.join(stray.to_string()).exists(),
|
||||
"degraded object's data dir must be preserved"
|
||||
);
|
||||
}
|
||||
|
||||
// Cross-replica union: a data dir referenced by ANOTHER disk's xl.meta must be
|
||||
// kept even where the local replica does not name it.
|
||||
#[tokio::test]
|
||||
async fn reclaim_orphan_data_dirs_keeps_dir_referenced_by_other_replica() {
|
||||
let (dir0, disk0) = make_single_local_disk().await;
|
||||
let (dir1, disk1) = make_single_local_disk().await;
|
||||
let shared = Uuid::new_v4();
|
||||
|
||||
// disk0: meta references only a different dir, but physically holds `shared`.
|
||||
let other = Uuid::new_v4();
|
||||
let obj0 = dir0.path().join("bucket").join("obj");
|
||||
write_object_meta_with_data_dirs(&obj0, "bucket", "obj", &[other]).await;
|
||||
fs::create_dir_all(obj0.join(other.to_string()))
|
||||
.await
|
||||
.expect("dir should be created");
|
||||
fs::create_dir_all(obj0.join(shared.to_string()))
|
||||
.await
|
||||
.expect("dir should be created");
|
||||
|
||||
// disk1: meta references `shared`.
|
||||
let obj1 = dir1.path().join("bucket").join("obj");
|
||||
write_object_meta_with_data_dirs(&obj1, "bucket", "obj", &[shared]).await;
|
||||
fs::create_dir_all(obj1.join(shared.to_string()))
|
||||
.await
|
||||
.expect("dir should be created");
|
||||
|
||||
let set = make_set_disks_with(vec![Some(disk0), Some(disk1)]).await;
|
||||
let removed = set
|
||||
.reclaim_orphan_data_dirs("bucket", "obj")
|
||||
.await
|
||||
.expect("reclaim should succeed");
|
||||
|
||||
assert_eq!(removed, 0, "a dir referenced by any replica must be kept");
|
||||
assert!(obj0.join(shared.to_string()).exists(), "cross-referenced data dir must survive");
|
||||
assert!(obj0.join(other.to_string()).exists(), "locally referenced data dir must survive");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial]
|
||||
async fn test_acquire_dist_delete_object_locks_batch_succeeds_with_two_healthy_lockers() {
|
||||
|
||||
@@ -624,6 +624,20 @@ impl SetDisks {
|
||||
|
||||
record_capacity_scope_if_needed(None, &out_dated_disks);
|
||||
|
||||
// The object is healthy here; sweep any data dirs left behind
|
||||
// by pre-#3510 unversioned overwrites, which the dangling paths
|
||||
// above never touch (issues #3231, #3191). Best effort — a
|
||||
// failure must not fail the heal.
|
||||
match self.reclaim_orphan_data_dirs(bucket, object).await {
|
||||
Ok(removed) if removed > 0 => {
|
||||
info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
|
||||
}
|
||||
}
|
||||
|
||||
Ok((result, None))
|
||||
}
|
||||
Err(err) => Ok((result, Some(err))),
|
||||
|
||||
Reference in New Issue
Block a user