diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 01171b8f1..940638853 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -13,9 +13,9 @@ // limitations under the License. use crate::disk::{ - CheckPartsResp, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint, Error, - FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result, - UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint, + Error, FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result, + SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, health_state::{ RuntimeDriveHealthState, classify_drive_recovery, get_drive_returning_probe_interval, get_drive_returning_success_threshold, get_drive_suspect_failure_threshold, record_drive_offline_duration, @@ -1349,6 +1349,30 @@ impl DiskAPI for LocalDiskWrapper { .await } + async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result { + self.track_disk_health( + || async { self.disk.acquire_snapshot_lease(volume, path).await }, + get_max_timeout_duration(), + ) + .await + } + + async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> { + self.track_disk_health( + || async { self.disk.release_snapshot_lease(volume, path, token).await }, + get_max_timeout_duration(), + ) + .await + } + + async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { + self.track_disk_health( + || async { self.disk.delete_data_dir(volume, path, opts).await }, + get_max_timeout_duration(), + ) + .await + } + async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { self.track_disk_health( || async { self.disk.write_metadata(org_volume, volume, path, fi).await }, diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index c24e38385..991a2248b 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -18,11 +18,12 @@ use crate::data_usage::local_snapshot::ensure_data_usage_layout; use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout}; use crate::disk::{ BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, - CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, - FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, PART_TRANSACTION_NEW_META, - PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, - RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, - STORAGE_FORMAT_FILE_BACKUP, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int, + CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, + DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, + PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET, + RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, + STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + conv_part_err_to_int, endpoint::Endpoint, error::{DiskError, Error, FileAccessDeniedWithContext, Result}, error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error}, @@ -66,7 +67,7 @@ use tokio::fs::{self, File}; #[cfg(not(unix))] use tokio::io::AsyncReadExt; use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf}; -use tokio::sync::{Notify, RwLock, Semaphore}; +use tokio::sync::{Mutex, Notify, RwLock, Semaphore}; use tokio::time::{Instant, Sleep, interval_at, timeout}; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -3856,6 +3857,25 @@ pub struct LocalDisk { exit_signal: Option>, io_backend: Arc, file_sync_permits: Arc, + snapshot_leases: Arc>, +} + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct SnapshotLeaseKey { + volume: String, + path: String, +} + +#[derive(Default)] +struct SnapshotLeaseEntry { + tokens: HashSet, + pending_delete: Option, + deleting: bool, +} + +#[derive(Default)] +struct SnapshotLeaseRegistry { + entries: HashMap, } impl Drop for LocalDisk { @@ -4092,6 +4112,7 @@ impl LocalDisk { exit_signal: None, io_backend: build_local_io_backend(root.clone()), file_sync_permits: os::disk_file_sync_limiter(&root), + snapshot_leases: Arc::new(Mutex::new(SnapshotLeaseRegistry::default())), }; let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| { log_startup_disk_error("get_disk_info", &root, err); @@ -4576,6 +4597,23 @@ impl LocalDisk { Ok(()) } + async fn delete_unleased(&self, volume: &str, path: &str, opt: &DeleteOptions) -> Result<()> { + let volume_dir = self.get_bucket_path(volume)?; + if !skip_access_checks(volume) + && let Err(e) = access(&volume_dir).await + { + return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); + } + + let file_path = self.get_object_path(volume, path)?; + check_path_length(file_path.to_string_lossy().as_ref())?; + self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate) + .await?; + // A deleted shard must not remain readable through the io_uring fd cache. + self.io_backend.invalidate_cached_fds_under(volume, path); + Ok(()) + } + #[tracing::instrument(level = "trace", skip_all)] #[async_recursion::async_recursion] async fn delete_file( @@ -6216,27 +6254,7 @@ impl DiskAPI for LocalDisk { #[tracing::instrument(level = "trace", skip_all)] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { crate::hp_guard!("LocalDisk::delete"); - let volume_dir = self.get_bucket_path(volume)?; - if !skip_access_checks(volume) - && let Err(e) = access(&volume_dir).await - { - return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); - } - - let file_path = self.get_object_path(volume, path)?; - - check_path_length(file_path.to_string_lossy().to_string().as_str())?; - - self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate) - .await?; - - // The inode is unlinked, but a cached descriptor would keep it readable — - // a deleted shard must not keep answering reads (backlog#1145). The part - // numbers under `path` are not known here, so this is the one caller that - // needs the predicate form. - self.io_backend.invalidate_cached_fds_under(volume, path); - - Ok(()) + self.delete_unleased(volume, path, &opt).await } #[tracing::instrument(level = "trace", skip_all)] @@ -7824,6 +7842,118 @@ impl DiskAPI for LocalDisk { Ok(()) } + async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result { + let file_path = self.get_object_path(volume, path)?; + let key = SnapshotLeaseKey { + volume: volume.to_string(), + path: path.to_string(), + }; + let token = { + let mut registry = self.snapshot_leases.lock().await; + if registry.entries.get(&key).is_some_and(|entry| entry.deleting) { + return Err(DiskError::FileNotFound); + } + let token = SnapshotLeaseToken::new(); + registry.entries.entry(key).or_default().tokens.insert(token); + token + }; + match fs::metadata(file_path).await { + Ok(metadata) if metadata.is_dir() => Ok(token), + Ok(_) => { + self.release_snapshot_lease(volume, path, token).await?; + Err(DiskError::FileNotFound) + } + Err(err) => { + self.release_snapshot_lease(volume, path, token).await?; + Err(to_file_error(err).into()) + } + } + } + + async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> { + let key = SnapshotLeaseKey { + volume: volume.to_string(), + path: path.to_string(), + }; + let opts = { + let mut registry = self.snapshot_leases.lock().await; + let Some(entry) = registry.entries.get_mut(&key) else { + return Ok(()); + }; + entry.tokens.remove(&token); + if !entry.tokens.is_empty() || entry.deleting { + return Ok(()); + } + + let Some(opts) = entry.pending_delete.clone() else { + registry.entries.remove(&key); + return Ok(()); + }; + entry.deleting = true; + opts + }; + let result = self.delete_unleased(volume, path, &opts).await; + let mut registry = self.snapshot_leases.lock().await; + match result { + Ok(()) => { + registry.entries.remove(&key); + Ok(()) + } + Err(err) => { + if let Some(entry) = registry.entries.get_mut(&key) { + entry.deleting = false; + } + Err(err) + } + } + } + + async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { + let key = SnapshotLeaseKey { + volume: volume.to_string(), + path: path.to_string(), + }; + { + let mut registry = self.snapshot_leases.lock().await; + if let Some(entry) = registry.entries.get_mut(&key) { + if !entry.tokens.is_empty() { + entry.pending_delete.get_or_insert_with(|| opts.clone()); + return Ok(DataDirDeleteStatus::Deferred); + } + if entry.deleting { + entry.pending_delete.get_or_insert_with(|| opts.clone()); + return Ok(DataDirDeleteStatus::Deferred); + } + entry.deleting = true; + entry.pending_delete.get_or_insert_with(|| opts.clone()); + } else { + registry.entries.insert( + key.clone(), + SnapshotLeaseEntry { + pending_delete: Some(opts.clone()), + deleting: true, + ..Default::default() + }, + ); + } + } + + let result = self.delete_unleased(volume, path, &opts).await; + let mut registry = self.snapshot_leases.lock().await; + match result { + Ok(()) => { + registry.entries.remove(&key); + Ok(DataDirDeleteStatus::Deleted) + } + Err(err) => { + if let Some(entry) = registry.entries.get_mut(&key) { + entry.deleting = false; + } + Err(err) + } + } + } + #[tracing::instrument(level = "trace", skip_all)] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { if !fi.metadata.is_empty() { @@ -14800,6 +14930,162 @@ mod test { assert!(construction_source.is::()); } + #[tokio::test] + async fn snapshot_leases_defer_data_dir_cleanup_until_last_release() { + use tempfile::tempdir; + + let root_dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let volume = "snapshot-lease-volume"; + let data_dir = path_join_buf(&["object", &Uuid::new_v4().to_string()]); + let first_part = path_join_buf(&[&data_dir, "part.1"]); + let later_part = path_join_buf(&[&data_dir, "part.2"]); + ensure_test_volume(&disk, volume).await; + disk.write_all(volume, &first_part, Bytes::from_static(b"first")) + .await + .expect("first shard should be written"); + disk.write_all(volume, &later_part, Bytes::from_static(b"later")) + .await + .expect("later shard should be written"); + + let first = disk + .acquire_snapshot_lease(volume, &data_dir) + .await + .expect("first lease should be acquired"); + let second = disk + .acquire_snapshot_lease(volume, &data_dir) + .await + .expect("second lease should be acquired"); + let status = disk + .delete_data_dir( + volume, + &data_dir, + DeleteOptions { + recursive: true, + ..Default::default() + }, + ) + .await + .expect("cleanup should be deferred"); + assert_eq!(status, DataDirDeleteStatus::Deferred); + assert_eq!( + disk.read_all(volume, &later_part) + .await + .expect("a later multipart shard must remain openable while leased"), + Bytes::from_static(b"later") + ); + + disk.release_snapshot_lease(volume, &data_dir, first) + .await + .expect("first lease release should succeed"); + assert!( + disk.read_all(volume, &first_part).await.is_ok(), + "one remaining lease must keep the data directory" + ); + disk.release_snapshot_lease(volume, &data_dir, second) + .await + .expect("last lease release should run deferred cleanup"); + disk.release_snapshot_lease(volume, &data_dir, second) + .await + .expect("releasing an already released token should be idempotent"); + assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound))); + } + + #[tokio::test] + async fn data_dir_cleanup_without_a_lease_keeps_existing_behavior() { + use tempfile::tempdir; + + let root_dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let volume = "snapshot-no-lease-volume"; + let data_dir = path_join_buf(&["object", &Uuid::new_v4().to_string()]); + let part = path_join_buf(&[&data_dir, "part.1"]); + ensure_test_volume(&disk, volume).await; + disk.write_all(volume, &part, Bytes::from_static(b"payload")) + .await + .expect("test shard should be written"); + + let status = disk + .delete_data_dir( + volume, + &data_dir, + DeleteOptions { + recursive: true, + ..Default::default() + }, + ) + .await + .expect("unleased cleanup should retain the existing delete behavior"); + assert_eq!(status, DataDirDeleteStatus::Deleted); + assert!(matches!(disk.read_all(volume, &part).await, Err(DiskError::FileNotFound))); + } + + #[tokio::test] + async fn snapshot_lease_acquire_and_cleanup_are_atomic() { + use tempfile::tempdir; + + let root_dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let volume = "snapshot-race-volume"; + ensure_test_volume(&disk, volume).await; + + for iteration in 0..32 { + let data_dir = path_join_buf(&["object", &format!("{iteration:032x}")]); + let part = path_join_buf(&[&data_dir, "part.1"]); + disk.write_all(volume, &part, Bytes::from_static(b"payload")) + .await + .expect("test shard should be written"); + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + let acquire_disk = Arc::clone(&disk); + let acquire_barrier = Arc::clone(&barrier); + let acquire_path = data_dir.clone(); + let acquire = tokio::spawn(async move { + acquire_barrier.wait().await; + acquire_disk.acquire_snapshot_lease(volume, &acquire_path).await + }); + let delete_disk = Arc::clone(&disk); + let delete_barrier = Arc::clone(&barrier); + let delete_path = data_dir.clone(); + let delete = tokio::spawn(async move { + delete_barrier.wait().await; + delete_disk + .delete_data_dir( + volume, + &delete_path, + DeleteOptions { + recursive: true, + ..Default::default() + }, + ) + .await + }); + barrier.wait().await; + + let acquired = acquire.await.expect("acquire task should join"); + let deleted = delete + .await + .expect("delete task should join") + .expect("delete should either run or defer"); + match acquired { + Ok(token) => { + assert_eq!(deleted, DataDirDeleteStatus::Deferred); + assert!(disk.read_all(volume, &part).await.is_ok()); + disk.release_snapshot_lease(volume, &data_dir, token) + .await + .expect("release should finish deferred cleanup"); + } + Err(DiskError::FileNotFound) => { + assert_eq!(deleted, DataDirDeleteStatus::Deleted); + } + Err(err) => panic!("unexpected lease acquisition error: {err}"), + } + assert!(matches!(disk.read_all(volume, &part).await, Err(DiskError::FileNotFound))); + } + } + #[tokio::test] async fn local_disk_check_parts_rejects_zero_data_geometry_before_shard_math() { use tempfile::tempdir; diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index fc17998e8..ee947d291 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -72,6 +72,27 @@ pub type DiskStore = Arc; pub type FileReader = Box; pub type FileWriter = Box; +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)] +pub struct SnapshotLeaseToken(Uuid); + +impl SnapshotLeaseToken { + pub fn new() -> Self { + Self(Uuid::new_v4()) + } +} + +impl Default for SnapshotLeaseToken { + fn default() -> Self { + Self::new() + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DataDirDeleteStatus { + Deleted, + Deferred, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum PartTransactionAction { Commit, @@ -249,6 +270,27 @@ impl DiskAPI for Disk { } } + async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result { + match self { + Disk::Local(local_disk) => local_disk.acquire_snapshot_lease(volume, path).await, + Disk::Remote(remote_disk) => remote_disk.acquire_snapshot_lease(volume, path).await, + } + } + + async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> { + match self { + Disk::Local(local_disk) => local_disk.release_snapshot_lease(volume, path, token).await, + Disk::Remote(remote_disk) => remote_disk.release_snapshot_lease(volume, path, token).await, + } + } + + async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { + match self { + Disk::Local(local_disk) => local_disk.delete_data_dir(volume, path, opts).await, + Disk::Remote(remote_disk) => remote_disk.delete_data_dir(volume, path, opts).await, + } + } + #[tracing::instrument(level = "trace", skip_all)] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { match self { @@ -646,6 +688,16 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { ) -> Result<()>; async fn delete_versions(&self, volume: &str, versions: Vec, opts: DeleteOptions) -> Vec>; async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()>; + async fn acquire_snapshot_lease(&self, _volume: &str, _path: &str) -> Result { + Err(Error::other("snapshot leases are not supported by this disk")) + } + async fn release_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<()> { + Err(Error::other("snapshot leases are not supported by this disk")) + } + async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { + self.delete(volume, path, opts).await?; + Ok(DataDirDeleteStatus::Deleted) + } async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()>; async fn read_version( diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index dd5a9b5b5..4972274f1 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -47,8 +47,8 @@ use crate::diagnostics::get::{ record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled, }; use crate::disk::{ - OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, - part_transaction_path, + DataDirDeleteStatus, OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, + PartTransactionAction, part_transaction_path, }; use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::ShardReader; @@ -2985,29 +2985,48 @@ impl SetDisks { Self::rename_fanout_barrier(&object_for_fault, idx, rename_fanout_barrier_phase::CLEANUP).await; if let Some(err) = Self::cleanup_injected_error(&object_for_fault, idx) { - return Some(err); + return (false, Some(err)); } if let Some(disk) = disk { - disk.delete( - &bucket, - &file_path, - DeleteOptions { - recursive: true, - ..Default::default() - }, - ) - .await - .err() + match disk + .delete_data_dir( + &bucket, + &file_path, + DeleteOptions { + recursive: true, + ..Default::default() + }, + ) + .await + { + Ok(DataDirDeleteStatus::Deleted) => (false, None), + Ok(DataDirDeleteStatus::Deferred) => (true, None), + Err(err) => (false, Some(err)), + } } else { // `None` slot: ignored placeholder. It is not `attempted`, so // classification excludes it from residue regardless. - Some(DiskError::DiskNotFound) + (false, Some(DiskError::DiskNotFound)) } }) }); - let errs: Vec> = join_all(futures).await.into_iter().map(map_cleanup_join_result).collect(); + let mut deferred = 0usize; + let errs: Vec> = join_all(futures) + .await + .into_iter() + .map(|result| match result { + Ok((was_deferred, err)) => { + deferred += usize::from(was_deferred); + err + } + Err(join_err) => Some(DiskError::other(format!("old data dir cleanup task failed: {join_err}"))), + }) + .collect(); - classify_old_data_dir_cleanup(&errs, &attempted, write_quorum) + let mut cleanup = classify_old_data_dir_cleanup(&errs, &attempted, write_quorum); + cleanup.deferred = deferred; + cleanup.reclaimed = cleanup.reclaimed.saturating_sub(deferred); + cleanup } /// Test-only fault-injection seam for the old-data-dir cleanup path @@ -3098,6 +3117,20 @@ impl SetDisks { rustfs_io_metrics::record_old_data_dir_cleanup(c.attempted, c.reclaimed, c.unreclaimed_disks.len(), c.below_quorum); + if c.deferred > 0 { + debug!( + event = EVENT_SET_DISK_WRITE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket = %bucket, + object = %object, + old_data_dir = %old_dir, + deferred = c.deferred, + state = "old_data_cleanup_deferred", + "Old data directory cleanup deferred for active snapshot leases" + ); + } + if actions.warn { warn!( component = LOG_COMPONENT_ECSTORE, @@ -4129,6 +4162,9 @@ pub(in crate::set_disk) struct OldDataDirCleanup { /// Number of attempted disks that returned `Ok` or a not-found variant /// (a missing dir == already reclaimed). pub reclaimed: usize, + /// Number of attempted disks that retained the directory for an active + /// snapshot lease and registered it for deletion after the final release. + pub deferred: usize, /// Indices of attempted disks that failed with a non-ignored, non-not-found /// error (including task panic/cancel). This is the residue that actually /// leaks and drives the leak metric + heal enqueue. @@ -4191,6 +4227,7 @@ fn classify_old_data_dir_cleanup(errs: &[Option], attempted: &[bool], OldDataDirCleanup { attempted: attempted_count, reclaimed, + deferred: 0, unreclaimed_disks, below_quorum, } @@ -5131,6 +5168,40 @@ mod tests { drop((disk1, disk2)); } + #[tokio::test] + async fn commit_cleanup_reports_and_releases_deferred_snapshot_data_dirs() { + let bucket = "cleanup-lease-bucket"; + let object = "cleanup-lease-object"; + let old_data_dir = "11111111-1111-1111-1111-111111111111"; + let committed_data_dir = "22222222-2222-2222-2222-222222222222"; + let data_dir_path = format!("{object}/{old_data_dir}"); + let shard_path = format!("{data_dir_path}/part.1"); + let (_dir1, disk1) = read_multiple_test_disk(bucket, &[(&shard_path, b"one".as_slice())]).await; + let set = io_primitives_test_set(vec![Some(disk1.clone())], 0).await; + let lease = disk1 + .acquire_snapshot_lease(bucket, &data_dir_path) + .await + .expect("snapshot lease should be acquired before cleanup"); + + let cleanup = set + .commit_rename_data_dir(&[Some(disk1.clone())], bucket, object, old_data_dir, committed_data_dir, 1) + .await; + assert_eq!(cleanup.attempted, 1); + assert_eq!(cleanup.reclaimed, 0); + assert_eq!(cleanup.deferred, 1); + assert!(cleanup.unreclaimed_disks.is_empty()); + disk1 + .read_all(bucket, &shard_path) + .await + .expect("deferred cleanup must leave later shard opens available"); + + disk1 + .release_snapshot_lease(bucket, &data_dir_path, lease) + .await + .expect("final lease release should reclaim the old data directory"); + assert!(matches!(disk1.read_all(bucket, &shard_path).await, Err(DiskError::FileNotFound))); + } + /// Isolation guard: an armed barrier / observed object only affects its own /// object. A fan-out for a different (unobserved, unarmed) object must not be /// paused and must not accrue any tracked task count — so concurrent tests