diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 4d60b652b..3a28b840f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -472,7 +472,7 @@ pub mod event { pub mod global { pub use crate::runtime::global::{ set_global_endpoints, set_global_region, set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, - update_erasure_type, + update_erasure_type, wait_for_detached_mutations, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index e2f3b6588..f1aa911bc 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -5652,8 +5652,10 @@ async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal( } else { None }; + // The delete owns the publication guard from here on, so a cancelled + // expiry worker cannot let table-bucket publication overtake it. let mut dobj = match api - .delete_object_with_tier_delete_journal(&oi.bucket, &encode_dir_object(&oi.name), opts) + .delete_object_with_tier_delete_journal_and_guards(&oi.bucket, &encode_dir_object(&oi.name), opts, [publication_guard]) .await { Ok(dobj) => dobj, diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 425aa2a74..06a186587 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -2311,7 +2311,7 @@ impl DiskAPI for LocalDiskWrapper { // Record operation start self.health.last_started.store(current_unix_nanos(), Ordering::Relaxed); - self.health.increment_waiting(); + let waiting_guard = self.health.waiting_guard(); let metric_waiting_guard = self.metrics.waiting_guard(); let started = Instant::now(); @@ -2320,7 +2320,7 @@ impl DiskAPI for LocalDiskWrapper { self.metrics.record_operation_latency("delete_versions", started.elapsed()); self.record_batch_delete_error_metrics(&result); - self.health.decrement_waiting(); + drop(waiting_guard); drop(metric_waiting_guard); let has_err = result.iter().any(|e| e.is_some()); if !has_err { @@ -3229,6 +3229,40 @@ mod tests { assert_eq!(snapshot.total_deletes, 1); } + // A batch delete dropped while the disk call is in flight (for example by a + // cancelled caller) must not leave the drive looking busy forever. + #[tokio::test] + async fn delete_versions_releases_waiting_slot_when_dropped() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + wrapper.make_volume("bucket").await.expect("volume should be created"); + let object_path = disk.get_object_path("bucket", "object").expect("object path"); + // Holding the object's metadata lease parks the disk call before it mutates anything. + let metadata_lease = crate::disk::os::acquire_metadata_mutation_lease(&object_path, None).await; + + let mut delete = Box::pin(wrapper.delete_versions( + "bucket", + vec![FileInfoVersions { + name: "object".to_string(), + versions: vec![FileInfo::default()], + ..Default::default() + }], + DeleteOptions::default(), + )); + assert!( + futures::poll!(delete.as_mut()).is_pending(), + "the disk call should wait for the metadata lease" + ); + assert_eq!(wrapper.health.waiting_count(), 1); + + drop(delete); + assert_eq!(wrapper.health.waiting_count(), 0, "a dropped delete must release its waiting slot"); + drop(metadata_lease); + } + #[tokio::test] async fn delete_versions_counts_faulty_drive_availability_rejection() { let dir = tempfile::tempdir().expect("temp dir should be created"); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index b567a62c8..be9497308 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -76,7 +76,7 @@ use futures::{StreamExt, TryStreamExt, future::BoxFuture, stream}; use metrics::counter; #[cfg(target_os = "linux")] use metrics::gauge; -use parking_lot::RwLock as ParkingLotRwLock; +use parking_lot::{Mutex as ParkingLotMutex, RwLock as ParkingLotRwLock}; use rustfs_filemeta::{ Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn, ValidationMode, get_file_info, read_xl_meta_no_data_sync, @@ -87,6 +87,7 @@ use rustfs_utils::path::{ GLOBAL_DIR_SUFFIX, GLOBAL_DIR_SUFFIX_WITH_SLASH, SLASH_SEPARATOR, clean, decode_dir_object, encode_dir_object, has_suffix, path_join, path_join_buf, }; +use std::collections::BTreeSet; use std::collections::HashMap; use std::collections::HashSet; use std::fmt::Debug; @@ -103,7 +104,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::{Mutex, Notify, RwLock, Semaphore}; +use tokio::sync::{Notify, RwLock, Semaphore}; use tokio::time::{Instant, Sleep, interval_at, timeout}; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -160,10 +161,15 @@ pub(crate) fn metadata_less_part_file(entry: &str) -> bool { .is_some_and(|part_number| part_number.parse::().is_ok_and(|part_number| part_number > 0)) } -fn is_delete_transaction_marker(entry: &str, prefix: &str) -> bool { +fn delete_transaction_marker_id(entry: &str, prefix: &str) -> Option { entry .strip_prefix(prefix) - .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok_and(|uuid| !uuid.is_nil())) + .and_then(|transaction| Uuid::parse_str(transaction).ok()) + .filter(|uuid| !uuid.is_nil()) +} + +fn is_delete_transaction_marker(entry: &str, prefix: &str) -> bool { + delete_transaction_marker_id(entry, prefix).is_some() } /// Whether a `list_dir` entry inside a UUID data dir is erasure data, a rollback @@ -836,6 +842,8 @@ const EVENT_DISK_LOCAL_URING_READ_BUDGET: &str = "disk_local_uring_read_budget"; const EVENT_DISK_LOCAL_URING_RESULT_BUDGET: &str = "disk_local_uring_result_budget"; const EVENT_DISK_LOCAL_DELETE_FAILED: &str = "disk_local_delete_failed"; const EVENT_DISK_LOCAL_DELETE_ROLLBACK_FAILED: &str = "disk_local_delete_rollback_failed"; +/// A delete retry found committed delete residue of this object and ran its cleanup pass. +const EVENT_DISK_LOCAL_DELETE_RESIDUE_FINISHED: &str = "disk_local_delete_residue_finished"; const EVENT_DISK_LOCAL_CHECK_PARTS: &str = "disk_local_check_parts"; const EVENT_DISK_LOCAL_ACCESS_FAILED: &str = "disk_local_access_failed"; const EVENT_DISK_LOCAL_VOLUME_SETUP_FAILED: &str = "disk_local_volume_setup_failed"; @@ -844,6 +852,9 @@ const EVENT_DISK_LOCAL_FORMAT_DECODE_FAILED: &str = "disk_local_format_decode_fa /// to replace. Best effort — the rename that follows fails closed — but a /// recurring signal means heal is stuck on that drive. const EVENT_DISK_LOCAL_HEAL_PURGE_FAILED: &str = "disk_local_heal_purge_failed"; +/// Delete transactions whose committed residue a repeated delete finished on +/// a local disk (#6898), counted instead of logged per object. +const METRIC_DISK_DELETE_RESIDUE_FINISHED_TOTAL: &str = "rustfs_disk_delete_residue_finished_total"; #[cfg(unix)] const METRIC_GET_OBJECT_MMAP_PAGE_FAULTS_TOTAL: &str = "rustfs_io_get_object_mmap_page_faults_total"; #[cfg(unix)] @@ -5190,7 +5201,7 @@ pub struct LocalDisk { exit_signal: Option>, io_backend: Arc, file_sync_permits: Arc, - snapshot_leases: Arc>, + snapshot_leases: Arc>, } #[derive(Clone, Debug, Eq, Hash, PartialEq)] @@ -5230,6 +5241,47 @@ struct SnapshotLeaseRegistry { entries: HashMap, } +/// Owns the `deleting` flag of one registry entry while its physical delete +/// runs. A delete future dropped before it finishes (a timeout or a cancelled +/// caller) clears the flag, so a later lease release or volume settle retries +/// the pending delete instead of skipping the entry forever. +struct SnapshotLeaseDeleteClaim<'a> { + registry: &'a ParkingLotMutex, + key: SnapshotLeaseKey, + finished: bool, +} + +impl<'a> SnapshotLeaseDeleteClaim<'a> { + /// The caller must have set `deleting` on `key` under the registry lock. + fn new(registry: &'a ParkingLotMutex, key: SnapshotLeaseKey) -> Self { + Self { + registry, + key, + finished: false, + } + } + + fn finish(mut self, deleted: bool) { + self.finished = true; + let mut registry = self.registry.lock(); + if deleted { + registry.entries.remove(&self.key); + } else if let Some(entry) = registry.entries.get_mut(&self.key) { + entry.deleting = false; + } + } +} + +impl Drop for SnapshotLeaseDeleteClaim<'_> { + fn drop(&mut self) { + if !self.finished + && let Some(entry) = self.registry.lock().entries.get_mut(&self.key) + { + entry.deleting = false; + } + } +} + impl Drop for LocalDisk { fn drop(&mut self) { if let Some(exit_signal) = self.exit_signal.take() { @@ -5645,7 +5697,7 @@ impl LocalDisk { exit_signal: None, io_backend: build_local_io_backend(io_root.clone()).await, file_sync_permits: os::disk_file_sync_limiter(&root), - snapshot_leases: Arc::new(Mutex::new(SnapshotLeaseRegistry::default())), + snapshot_leases: Arc::new(ParkingLotMutex::new(SnapshotLeaseRegistry::default())), }; let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| { log_startup_disk_error("get_disk_info", &root, err); @@ -6216,7 +6268,7 @@ impl LocalDisk { path: path.to_string(), }; { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); if let Some(entry) = registry.entries.get_mut(&key) { if !entry.tokens.is_empty() { entry.pending_delete.get_or_insert_with(|| opts.clone()); @@ -6239,23 +6291,13 @@ impl LocalDisk { ); } } + let claim = SnapshotLeaseDeleteClaim::new(&self.snapshot_leases, key); let result = self .delete_unleased_with_namespace_owner(volume, path, &opts, namespace_owner) .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) - } - } + claim.finish(result.is_ok()); + result.map(|()| DataDirDeleteStatus::Deleted) } async fn delete_version_inner(&self, volume: &str, path: &str, fi: FileInfo, mutation: DeleteVersionMutation) -> Result<()> { @@ -6318,6 +6360,8 @@ impl LocalDisk { if err != DiskError::FileNotFound { return Err(err); } + self.finish_committed_delete_residue(volume, path, namespace_owner.clone()) + .await; if fi.deleted && force_del_marker { return self @@ -7279,6 +7323,8 @@ impl LocalDisk { { Ok(data) => data, Err(DiskError::FileNotFound) => { + self.finish_committed_delete_residue(volume, path, namespace_owner.clone()) + .await; // `deleted` alone can be an explicit marker purge; only // `mark_deleted` may create metadata that was not present. let Some(delete_marker) = fis.iter().find(|fi| fi.deleted && fi.mark_deleted).cloned() else { @@ -9101,7 +9147,7 @@ impl LocalDisk { path: quota_mutation_fence_path(volume, path), }; let state = { - let registry = self.snapshot_leases.lock().await; + let registry = self.snapshot_leases.lock(); let entry = registry.entries.get(&key).ok_or(DiskError::FileNotFound)?; let state = entry.mutation_fence.as_ref().ok_or(DiskError::FileNotFound)?; if !entry.tokens.contains(&token) || state.revoked.load(Ordering::Acquire) { @@ -9216,6 +9262,150 @@ impl LocalDisk { Ok(()) } + /// Run the cleanup pass of delete transactions that committed on this disk + /// but never cleaned up, for example because an older build stopped the + /// delete when its client disconnected (#6898). Returns how many were + /// finished. + /// + /// Called only where `path` was just found without `xl.meta`, under the + /// object's metadata lease. A `delete-data.` marker proves that `T` + /// removed the object's metadata on this disk and committed, so the data + /// it marked can no longer be read, healed or rolled back here. Finishing + /// removes what `T`'s own cleanup pass would have removed. Reserved + /// markers, rollback backups without a committed marker, and object + /// directories that hold `xl.meta` are left alone. + async fn finish_committed_delete_residue( + &self, + volume: &str, + path: &str, + namespace_owner: Option>, + ) -> usize { + let transactions = match self.committed_delete_transactions(volume, path).await { + Ok(transactions) => transactions, + Err(err) => { + warn!( + event = EVENT_DISK_LOCAL_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + path, + operation = "scan_committed_delete_residue", + error = ?err, + "Disk local delete failed" + ); + return 0; + } + }; + let opts = DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }; + let mut finished = 0; + for transaction in transactions { + // Unlike the delete's own pass, drop the rollback backup before the + // marked data: if this is cut short, the marker that proves the + // transaction is still in place for the next attempt. + let result = match self + .delete_unleased_with_namespace_owner(volume, &format!("{path}/{transaction}"), &opts, namespace_owner.clone()) + .await + { + Ok(()) | Err(DiskError::FileNotFound) => self + .finish_version_delete(volume, path, transaction, namespace_owner.clone()) + .await + .map(|_| ()), + Err(err) => Err(err), + }; + match result { + Ok(()) => finished += 1, + Err(err) => warn!( + event = EVENT_DISK_LOCAL_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + path, + transaction = %transaction, + operation = "finish_committed_delete_residue", + error = ?err, + "Disk local delete failed" + ), + } + } + if finished > 0 { + counter!(METRIC_DISK_DELETE_RESIDUE_FINISHED_TOTAL).increment(u64::try_from(finished).unwrap_or(u64::MAX)); + debug!( + event = EVENT_DISK_LOCAL_DELETE_RESIDUE_FINISHED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + path, + state = "finished", + transactions = finished, + "Disk local delete residue finished" + ); + } + finished + } + + /// Transactions with a committed `delete-data.` marker in a data dir of + /// the metadata-less object directory `path`. Empty when the directory is + /// absent or holds anything but UUID data dirs, `xl.meta` included. + async fn committed_delete_transactions(&self, volume: &str, path: &str) -> Result> { + let object_path = self.io_get_object_path(volume, path)?; + let mut entries = match fs::read_dir(&object_path).await { + Ok(entries) => entries, + Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => return Ok(BTreeSet::new()), + Err(err) => return Err(to_file_error(err).into()), + }; + let mut data_dirs = Vec::new(); + while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? { + // Residue holds only UUID data dirs: `xl.meta` is live metadata and + // any other child makes `path` a prefix of other keys. Stop at the + // first such child instead of walking a possibly large prefix. + let is_data_dir_name = entry + .file_name() + .to_str() + .is_some_and(|name| Uuid::parse_str(name).is_ok_and(|uuid| !uuid.is_nil())); + if !is_data_dir_name { + return Ok(BTreeSet::new()); + } + match entry.file_type().await { + Ok(file_type) if file_type.is_dir() => data_dirs.push(entry.path()), + Ok(_) => return Ok(BTreeSet::new()), + // Removed since the directory was read, e.g. by a racing purge. + Err(err) if err.kind() == ErrorKind::NotFound => {} + Err(err) => return Err(to_file_error(err).into()), + } + } + + let mut transactions = BTreeSet::new(); + for data_dir in data_dirs { + let mut entries = match fs::read_dir(&data_dir).await { + Ok(entries) => entries, + Err(err) if err.kind() == ErrorKind::NotFound => continue, + Err(err) => return Err(to_file_error(err).into()), + }; + while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? { + let Some(transaction) = entry + .file_name() + .to_str() + .and_then(|name| delete_transaction_marker_id(name, DELETE_DATA_DIR_MARKER_PREFIX)) + else { + continue; + }; + match entry.file_type().await { + Ok(file_type) if file_type.is_file() => { + transactions.insert(transaction); + } + Ok(_) => {} + Err(err) if err.kind() == ErrorKind::NotFound => {} + Err(err) => return Err(to_file_error(err).into()), + } + } + } + Ok(transactions) + } + async fn finish_version_delete( &self, volume: &str, @@ -9293,7 +9483,7 @@ impl LocalDisk { /// only path-based reopens observe the removal. async fn settle_pending_snapshot_deletes(&self, volume: &str) { let pending: Vec<(SnapshotLeaseKey, DeleteOptions)> = { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); registry .entries .iter_mut() @@ -9304,26 +9494,22 @@ impl LocalDisk { }) .collect() }; + let pending: Vec<(SnapshotLeaseDeleteClaim<'_>, DeleteOptions)> = pending + .into_iter() + .map(|(key, opts)| (SnapshotLeaseDeleteClaim::new(&self.snapshot_leases, key), opts)) + .collect(); - for (key, opts) in pending { - let result = self.delete_unleased(&key.volume, &key.path, &opts).await; - let mut registry = self.snapshot_leases.lock().await; - match result { - Ok(()) => { - registry.entries.remove(&key); - } - Err(err) => { - if let Some(entry) = registry.entries.get_mut(&key) { - entry.deleting = false; - } - warn!( - volume = %key.volume, - path = %key.path, - error = %err, - "failed to settle deferred data-dir deletion before volume removal" - ); - } + for (claim, opts) in pending { + let result = self.delete_unleased(&claim.key.volume, &claim.key.path, &opts).await; + if let Err(err) = &result { + warn!( + volume = %claim.key.volume, + path = %claim.key.path, + error = %err, + "failed to settle deferred data-dir deletion before volume removal" + ); } + claim.finish(result.is_ok()); } } } @@ -10717,7 +10903,7 @@ impl DiskAPI for LocalDisk { path: path.to_string(), }; if volume == RUSTFS_META_BUCKET && is_quota_mutation_fence_path(path) { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); let entry = registry.entries.entry(key).or_default(); let state = entry .mutation_fence @@ -10733,7 +10919,7 @@ impl DiskAPI for LocalDisk { let file_path = self.io_get_object_path(volume, path)?; let _mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, volume, &file_path).await; let token = { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); if registry.entries.get(&key).is_some_and(|entry| entry.deleting) { return Err(DiskError::FileNotFound); } @@ -10761,7 +10947,7 @@ impl DiskAPI for LocalDisk { }; if volume == RUSTFS_META_BUCKET && is_quota_mutation_fence_path(path) { if !token.is_revoke_all() { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); let Some(entry) = registry.entries.get_mut(&key) else { return Ok(()); }; @@ -10777,7 +10963,7 @@ impl DiskAPI for LocalDisk { return Ok(()); } let state = { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); let Some(entry) = registry.entries.get_mut(&key) else { return Ok(()); }; @@ -10798,11 +10984,11 @@ impl DiskAPI for LocalDisk { } notified.await; } - self.snapshot_leases.lock().await.entries.remove(&key); + self.snapshot_leases.lock().entries.remove(&key); return Ok(()); } let opts = { - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); let Some(entry) = registry.entries.get_mut(&key) else { return Ok(()); }; @@ -10818,20 +11004,10 @@ impl DiskAPI for LocalDisk { entry.deleting = true; opts }; + let claim = SnapshotLeaseDeleteClaim::new(&self.snapshot_leases, key); 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) - } - } + claim.finish(result.is_ok()); + result } async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result { @@ -10839,7 +11015,7 @@ impl DiskAPI for LocalDisk { volume: volume.to_string(), path: path.to_string(), }; - let mut registry = self.snapshot_leases.lock().await; + let mut registry = self.snapshot_leases.lock(); let Some(entry) = registry.entries.get_mut(&key) else { return Err(DiskError::FileNotFound); }; @@ -15264,7 +15440,6 @@ mod test { let fence = disk .snapshot_leases .lock() - .await .entries .get(&SnapshotLeaseKey { volume: RUSTFS_META_BUCKET.to_string(), @@ -18060,6 +18235,326 @@ mod test { assert!(!object_dir.join(rollback_dir.to_string()).exists()); } + struct CommittedDeleteResidue { + object_dir: PathBuf, + data_dir: Uuid, + transaction: Uuid, + version_id: Uuid, + } + + impl CommittedDeleteResidue { + fn part(&self) -> PathBuf { + self.object_dir.join(self.data_dir.to_string()).join("part.1") + } + + fn marker(&self) -> PathBuf { + self.object_dir + .join(self.data_dir.to_string()) + .join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", self.transaction)) + } + + fn backup(&self) -> PathBuf { + self.object_dir + .join(self.transaction.to_string()) + .join(STORAGE_FORMAT_FILE_BACKUP) + } + } + + /// Leave the state a delete reaches once it committed on this disk but + /// before its cleanup pass: what a cancelled request used to strand. + async fn write_committed_delete_residue(disk: &LocalDisk, bucket: &str, object: &str) -> CommittedDeleteResidue { + let version_id = Uuid::new_v4(); + let data_dir = Uuid::new_v4(); + let transaction = Uuid::new_v4(); + let object_dir = disk.io_get_object_path(bucket, object).expect("object IO path"); + fs::create_dir_all(object_dir.join(data_dir.to_string())) + .await + .expect("data dir should be created"); + fs::write(object_dir.join(data_dir.to_string()).join("part.1"), b"stale") + .await + .expect("part data should be written"); + let fi = test_file_info(object, version_id, Some(data_dir), None); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), test_meta(fi.clone())) + .await + .expect("metadata should be written"); + disk.delete_version( + bucket, + object, + fi, + false, + DeleteOptions { + old_data_dir: Some(transaction), + ..Default::default() + }, + ) + .await + .expect("the delete should commit on this disk"); + + let residue = CommittedDeleteResidue { + object_dir, + data_dir, + transaction, + version_id, + }; + assert!(!residue.object_dir.join(STORAGE_FORMAT_FILE).exists()); + assert!(residue.part().exists() && residue.marker().exists() && residue.backup().exists()); + residue + } + + fn residue_delete_request(object: &str) -> Vec { + vec![FileInfoVersions { + name: object.to_string(), + versions: vec![FileInfo { + name: object.to_string(), + ..Default::default() + }], + ..Default::default() + }] + } + + // #6898: repeating the delete of a key whose earlier delete committed but + // never cleaned up reclaims the residue and still reports the key missing. + #[tokio::test] + async fn delete_versions_finishes_committed_delete_residue() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "prefix/object"; + ensure_test_volume(&disk, bucket).await; + let residue = write_committed_delete_residue(&disk, bucket, object).await; + + let errors = disk + .delete_versions(bucket, residue_delete_request(object), DeleteOptions::default()) + .await; + + assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound)]), "{errors:?}"); + assert!(!residue.object_dir.exists(), "the committed residue must be reclaimed"); + assert!(!dir.path().join(bucket).join("prefix").exists(), "empty parents must be pruned"); + assert!(dir.path().join(bucket).exists()); + } + + #[tokio::test] + async fn delete_version_finishes_committed_delete_residue() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "object"; + ensure_test_volume(&disk, bucket).await; + let residue = write_committed_delete_residue(&disk, bucket, object).await; + + let err = disk + .delete_version( + bucket, + object, + test_file_info(object, residue.version_id, Some(residue.data_dir), None), + false, + DeleteOptions::default(), + ) + .await + .expect_err("the version is already deleted"); + + assert_eq!(err, DiskError::FileVersionNotFound); + assert!(!residue.object_dir.exists(), "the committed residue must be reclaimed"); + } + + // The rollback backup may already be gone; the committed marker alone + // proves the delete. + #[tokio::test] + async fn delete_versions_finishes_committed_marker_without_rollback_backup() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "object"; + ensure_test_volume(&disk, bucket).await; + let residue = write_committed_delete_residue(&disk, bucket, object).await; + fs::remove_dir_all(residue.object_dir.join(residue.transaction.to_string())) + .await + .expect("rollback dir should be removed"); + + let errors = disk + .delete_versions(bucket, residue_delete_request(object), DeleteOptions::default()) + .await; + + assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound)]), "{errors:?}"); + assert!(!residue.object_dir.exists(), "the committed marker alone must be enough"); + } + + // Only a committed marker on this disk proves the delete. A reserved marker, + // a rollback backup alone, or live metadata beside the marker is left for + // the owning transaction, rollback, or heal. + #[tokio::test] + async fn committed_delete_residue_finish_leaves_unproven_state_alone() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + ensure_test_volume(&disk, bucket).await; + + // A rollback backup without a committed marker, as an inline object leaves it. + let backup_only = write_committed_delete_residue(&disk, bucket, "backup-only").await; + fs::remove_dir_all(backup_only.object_dir.join(backup_only.data_dir.to_string())) + .await + .expect("data dir should be removed"); + + // A delete that reserved its data dir but never removed the metadata. + let reserved = write_committed_delete_residue(&disk, bucket, "reserved").await; + fs::rename( + reserved.marker(), + reserved + .object_dir + .join(reserved.data_dir.to_string()) + .join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{}", reserved.transaction)), + ) + .await + .expect("marker should become a reservation"); + + // A committed marker beside live metadata. + let live = write_committed_delete_residue(&disk, bucket, "live").await; + fs::write(live.object_dir.join(STORAGE_FORMAT_FILE), b"live metadata") + .await + .expect("metadata should be written"); + + for object in ["backup-only", "reserved", "live"] { + assert_eq!( + disk.finish_committed_delete_residue(bucket, object, None).await, + 0, + "{object} carries no committed proof" + ); + } + let errors = disk + .delete_versions( + bucket, + [residue_delete_request("backup-only"), residue_delete_request("reserved")].concat(), + DeleteOptions::default(), + ) + .await; + assert!(errors.iter().all(|err| matches!(err, Some(DiskError::FileNotFound))), "{errors:?}"); + + assert!(backup_only.backup().exists()); + assert!(reserved.backup().exists() && reserved.part().exists()); + assert!(live.object_dir.join(STORAGE_FORMAT_FILE).exists()); + assert!(live.marker().exists() && live.part().exists() && live.backup().exists()); + } + + // A key that is also a prefix of other keys may name a huge directory; the + // first child that is not a data dir ends the scan and nothing is touched. + #[tokio::test] + async fn committed_delete_residue_finish_skips_a_prefix_of_other_keys() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "logs"; + ensure_test_volume(&disk, bucket).await; + let residue = write_committed_delete_residue(&disk, bucket, object).await; + let nested = residue.object_dir.join("2026").join(STORAGE_FORMAT_FILE); + fs::create_dir_all(nested.parent().expect("nested object dir")) + .await + .expect("nested object dir should be created"); + fs::write(&nested, b"nested metadata") + .await + .expect("nested metadata should be written"); + + assert_eq!(disk.finish_committed_delete_residue(bucket, object, None).await, 0); + assert!(residue.backup().exists() && residue.marker().exists() && residue.part().exists()); + assert!(nested.exists()); + } + + // A finish cut short before the marked data is gone (a timeout, a crash) + // must leave the marker that lets the next attempt recognize and finish + // the residue, never a lone rollback backup. + #[cfg(not(windows))] + #[tokio::test] + async fn committed_delete_residue_finish_removes_backup_before_its_proof() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "object"; + ensure_test_volume(&disk, bucket).await; + let residue = write_committed_delete_residue(&disk, bucket, object).await; + let data_path = disk + .io_get_object_path(bucket, &format!("{object}/{}", residue.data_dir)) + .expect("data dir IO path"); + let (destination_tx, destination_rx) = std::sync::mpsc::channel(); + let _destination = hooks::observe_rename_destination(&data_path, move |destination| { + let _ = destination_tx.send(destination.to_path_buf()); + }); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install_at(hooks::Stage::Rename, &data_path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + + let mut finish = Box::pin(disk.finish_committed_delete_residue(bucket, object, None)); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + entered = entered_rx => entered.expect("finish should reach the marked data dir"), + _ = finish.as_mut() => panic!("finish returned before removing the marked data dir"), + } + }) + .await + .expect("finish should start removing the marked data dir"); + + assert!(!residue.backup().exists(), "the rollback backup must go first"); + assert!(residue.marker().exists(), "the proof must outlive the backup"); + drop(finish); + + // The abandoned rename is still parked; the next attempt recognizes the + // residue by its marker and finishes it. + assert_eq!(disk.finish_committed_delete_residue(bucket, object, None).await, 1); + assert!(!residue.object_dir.exists(), "the retried finish must reclaim the residue"); + + // Let the abandoned rename run into its missing source and wait until + // it is done, so it cannot outlive the temp dir. + drop(release); + let destination = destination_rx + .try_recv() + .expect("the abandoned rename should have chosen its destination"); + tokio::time::timeout(Duration::from_secs(10), hooks::drain_namespace_key(&destination)) + .await + .expect("the abandoned rename should finish"); + } + + // Each transaction is finished only on its own proof. + #[tokio::test] + async fn committed_delete_residue_finish_is_per_transaction() { + let recorder = crate::test_metrics::CapturingRecorder::default(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let bucket = "bucket"; + let object = "object"; + ensure_test_volume(&disk, bucket).await; + let committed = write_committed_delete_residue(&disk, bucket, object).await; + fs::remove_dir_all(committed.object_dir.join(committed.transaction.to_string())) + .await + .expect("rollback dir of the committed transaction should be removed"); + let unproven = Uuid::new_v4(); + let unproven_backup = committed + .object_dir + .join(unproven.to_string()) + .join(STORAGE_FORMAT_FILE_BACKUP); + fs::create_dir_all(unproven_backup.parent().expect("backup parent")) + .await + .expect("unproven rollback dir should be created"); + fs::write(&unproven_backup, b"rollback metadata") + .await + .expect("unproven backup should be written"); + + assert_eq!(disk.finish_committed_delete_residue(bucket, object, None).await, 1); + assert!(!committed.object_dir.join(committed.data_dir.to_string()).exists()); + assert!(unproven_backup.exists(), "a backup without its own committed marker must remain"); + assert_eq!(recorder.counter_value(METRIC_DISK_DELETE_RESIDUE_FINISHED_TOTAL, &[]), 1); + } + #[tokio::test] async fn test_delete_version_error_after_staging_restores_data_dir() { use tempfile::tempdir; @@ -22390,6 +22885,79 @@ mod test { assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound))); } + // A data-dir delete dropped mid-flight (a DiskStore timeout or a cancelled + // caller) must hand its cleanup back instead of staying marked as deleting. + #[cfg(not(windows))] + #[tokio::test] + async fn cancelled_data_dir_delete_releases_its_snapshot_lease_claim() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + + let root_dir = tempfile::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-cancel"; + let data_dir = path_join_buf(&["object", &Uuid::new_v4().to_string()]); + ensure_test_volume(&disk, volume).await; + disk.write_all(volume, &path_join_buf(&[&data_dir, "part.1"]), Bytes::from_static(b"part")) + .await + .expect("shard should be written"); + let key = SnapshotLeaseKey { + volume: volume.to_string(), + path: data_dir.clone(), + }; + let data_path = disk.io_get_object_path(volume, &data_dir).expect("data dir IO path"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install_at(hooks::Stage::Rename, &data_path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + + let mut delete = Box::pin(disk.delete_data_dir( + volume, + &data_dir, + DeleteOptions { + recursive: true, + ..Default::default() + }, + )); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + entered = entered_rx => entered.expect("delete should reach its physical rename"), + _ = delete.as_mut() => panic!("delete returned before its physical rename"), + } + }) + .await + .expect("delete should start its physical rename"); + assert!( + disk.snapshot_leases + .lock() + .entries + .get(&key) + .is_some_and(|entry| entry.deleting) + ); + + drop(delete); + assert!( + disk.snapshot_leases + .lock() + .entries + .get(&key) + .is_some_and(|entry| !entry.deleting && entry.pending_delete.is_some()), + "a dropped delete must hand its pending cleanup back to the registry" + ); + drop(release); + disk.settle_pending_snapshot_deletes(volume).await; + assert!( + !disk.snapshot_leases.lock().entries.contains_key(&key), + "the handed-back cleanup must be retried" + ); + assert!(matches!( + disk.read_all(volume, &path_join_buf(&[&data_dir, "part.1"])).await, + Err(DiskError::FileNotFound) + )); + } + #[tokio::test] async fn quota_mutation_fence_revoke_waits_for_active_claim_and_rejects_late_claims() { use tempfile::tempdir; diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index 56ba22a0a..89312c135 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -307,6 +307,13 @@ pub fn shutdown_background_services() { } } +/// Wait up to `timeout` for object mutations that outlived their callers +/// (see `InstanceContext::wait_for_detached_mutations`). Returns `false` when +/// some are still running at the deadline. +pub async fn wait_for_detached_mutations(timeout: std::time::Duration) -> bool { + current_ctx().wait_for_detached_mutations(timeout).await +} + /// Set the global lock client (first LocalClient created) /// /// # Arguments diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 4c6193c43..d5c575ef0 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -59,6 +59,8 @@ use std::sync::{ use tokio::sync::{Mutex, Notify, OnceCell, OwnedRwLockReadGuard, RwLock}; use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; +use tracing::Instrument as _; use uuid::Uuid; const SCANNER_PUBLICATION_STATE_UNKNOWN: u8 = 0; @@ -243,6 +245,10 @@ pub struct InstanceContext { suppress_tier_delete_journal_recovery: bool, transition_transaction_recovery_stores: std::sync::Mutex>, tier_delete_journal_recovery_wakeup: tokio::sync::Notify, + /// Object mutations that run on their own task so that dropping the + /// caller (for example, an S3 client disconnect) cannot stop them between + /// their on-disk steps. Shutdown waits for them after the HTTP drain. + detached_mutations: TaskTracker, } impl InstanceContext { @@ -293,6 +299,7 @@ impl InstanceContext { suppress_tier_delete_journal_recovery: false, transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()), tier_delete_journal_recovery_wakeup: tokio::sync::Notify::new(), + detached_mutations: TaskTracker::new(), } } @@ -306,6 +313,31 @@ impl InstanceContext { self.lock_manager.clone() } + /// Run a multi-step mutation on its own task and wait for its output. + /// + /// Dropping the returned future leaves the task running to completion, so + /// the mutation, together with the locks and guards it owns, never stops + /// between two of its on-disk steps because its caller went away. + pub(crate) async fn run_detached_mutation(&self, mutation: F) -> Result + where + F: std::future::Future + Send + 'static, + F::Output: Send + 'static, + { + self.detached_mutations.spawn(mutation.in_current_span()).await + } + + /// Wait up to `timeout` for every detached mutation to finish. Returns + /// `false` when some are still running at the deadline. + pub async fn wait_for_detached_mutations(&self, timeout: std::time::Duration) -> bool { + self.detached_mutations.close(); + tokio::time::timeout(timeout, self.detached_mutations.wait()).await.is_ok() + } + + #[cfg(test)] + pub(crate) fn detached_mutation_count(&self) -> usize { + self.detached_mutations.len() + } + pub(crate) fn data_movement_operation_gate(&self) -> Arc> { Arc::clone(&self.data_movement_operation_gate) } @@ -920,6 +952,33 @@ mod tests { } } + // A detached mutation keeps running after its caller is dropped, and + // shutdown waits for it instead of tearing it down mid-way. + #[tokio::test] + async fn detached_mutation_outlives_dropped_caller_and_is_awaited() { + let ctx = InstanceContext::new(); + let finished = Arc::new(AtomicBool::new(false)); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); + let mut caller = Box::pin(ctx.run_detached_mutation({ + let finished = Arc::clone(&finished); + async move { + let _ = release_rx.await; + finished.store(true, Ordering::Release); + } + })); + assert!(futures::poll!(caller.as_mut()).is_pending()); + drop(caller); + assert_eq!(ctx.detached_mutation_count(), 1, "dropping the caller must not cancel the mutation"); + + assert!( + !ctx.wait_for_detached_mutations(std::time::Duration::from_millis(50)).await, + "the wait must report a mutation that is still running" + ); + release_tx.send(()).expect("the detached mutation should still be waiting"); + assert!(ctx.wait_for_detached_mutations(std::time::Duration::from_secs(5)).await); + assert!(finished.load(Ordering::Acquire)); + } + // A fresh context (before any update) reflects the initial all-false state. #[tokio::test] async fn fresh_context_is_all_false() { diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index e21c85484..14d69aaaa 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -2340,7 +2340,8 @@ enum OrphanDirScan { /// under taking parents, root included when anything under it is blocked); /// `dirs` is the pre-order list of every directory reached that is not /// blocked, and `committed_files` the erasure data and committed delete - /// markers found in the unblocked UUID data dirs among them. + /// markers found in the unblocked UUID data dirs among them, plus the + /// rollback backups those markers prove. Scanned { blocked: HashSet, dirs: Vec, @@ -2382,10 +2383,15 @@ fn is_safe_orphan_dir_entry(entry: &str) -> bool { && !component.contains('\\') } -fn is_committed_delete_marker(entry: &str) -> bool { +fn committed_delete_marker_transaction(entry: &str) -> Option { entry .strip_prefix(DELETE_DATA_DIR_MARKER_PREFIX) - .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok_and(|uuid| !uuid.is_nil())) + .and_then(|transaction| Uuid::parse_str(transaction).ok()) + .filter(|uuid| !uuid.is_nil()) +} + +fn is_committed_delete_marker(entry: &str) -> bool { + committed_delete_marker_transaction(entry).is_some() } /// Outcome of a *post-quorum* `rename_data` commit, classifying whether the @@ -5149,8 +5155,9 @@ impl SetDisks { } /// Scan a single disk's copy of `prefix` and classify every directory - /// under it. Only empty directories and UUID data directories with valid - /// committed delete markers are purgeable; anything else blocks its whole + /// under it. Only empty directories, UUID data directories with valid + /// committed delete markers, and the rollback backup of a delete whose + /// marker sits beside it are purgeable; anything else blocks its whole /// ancestor chain while sibling subtrees stay purgeable, so committed /// residue is still reclaimed when it shares an ancestor with residue from /// an older build that never wrote markers (#6898). @@ -5163,6 +5170,12 @@ impl SetDisks { let mut committed_files: Vec = Vec::new(); let mut blocked: HashSet = HashSet::new(); let mut existed = false; + // `/` directories holding only a delete's rollback backup, + // and the `(object, T)` pairs a committed `delete-data.` marker in a + // purgeable data dir of that object proves. Both are resolved after the + // walk because siblings are visited in no particular order. + let mut rollback_backup_dirs: Vec = Vec::new(); + let mut committed_transactions: HashSet<(String, Uuid)> = HashSet::new(); while let Some(dir) = stack.pop() { let entries = match disk.list_dir("", bucket, &dir, 0).await { @@ -5196,10 +5209,24 @@ impl SetDisks { } if !has_data && !files.is_empty() { - let data_dir_name = dir.rsplit(SLASH_SEPARATOR).next().unwrap_or_default(); + let (parent, data_dir_name) = dir.rsplit_once(SLASH_SEPARATOR).unwrap_or(("", dir.as_str())); let is_uuid_data_dir = Uuid::parse_str(data_dir_name).is_ok_and(|uuid| !uuid.is_nil()); - let has_committed_delete = files.iter().any(|entry| is_committed_delete_marker(entry)); - has_data = !is_uuid_data_dir || !has_committed_delete || files.iter().any(|entry| entry == STORAGE_FORMAT_FILE); + let committed = files + .iter() + .filter_map(|entry| committed_delete_marker_transaction(entry)) + .collect::>(); + if is_uuid_data_dir + && dir != root + && child_dirs.is_empty() + && matches!(files.as_slice(), [entry] if entry == STORAGE_FORMAT_FILE_BACKUP) + { + rollback_backup_dirs.push(dir); + continue; + } + has_data = !is_uuid_data_dir || committed.is_empty() || files.iter().any(|entry| entry == STORAGE_FORMAT_FILE); + if !has_data { + committed_transactions.extend(committed.into_iter().map(|transaction| (parent.to_string(), transaction))); + } } if has_data { @@ -5214,6 +5241,22 @@ impl SetDisks { stack.extend(child_dirs); } + // A rollback backup is purgeable only when a data dir of the same + // object proves its transaction committed on this disk; otherwise it may + // still be needed to roll the delete back, so it blocks as before. + for dir in rollback_backup_dirs { + let proven = dir.rsplit_once(SLASH_SEPARATOR).is_some_and(|(parent, transaction)| { + Uuid::parse_str(transaction) + .is_ok_and(|transaction| committed_transactions.contains(&(parent.to_string(), transaction))) + }); + if proven { + committed_files.push(format!("{dir}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE_BACKUP}")); + dirs.push(dir); + } else { + block_orphan_dir_chain(&mut blocked, &root, &dir); + } + } + if existed { OrphanDirScan::Scanned { blocked, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index ba79bb254..e184df9f8 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -885,8 +885,6 @@ pub(crate) use ops::hermetic_set_disks_isolated; pub(crate) use ops::multipart::NewMultipartUploadCommitObservation; #[cfg(any(test, feature = "test-util"))] pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause}; -#[cfg(test)] -pub(crate) use ops::object::DeleteObjectCommitBarrier; #[cfg(feature = "test-util")] pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier; #[cfg(all(test, feature = "test-util"))] @@ -894,6 +892,8 @@ pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransition pub(crate) use ops::object::body_cache_plaintext_len; #[cfg(all(test, feature = "test-util"))] pub(crate) use ops::object::cleanup_rejected_transition_upload_durably; +#[cfg(test)] +pub(crate) use ops::object::{DeleteCleanupBarrier, DeleteObjectCommitBarrier}; #[cfg(any(test, feature = "test-util"))] pub use ops::object::{PutObjectCommitBarrier, PutObjectCommitPause}; #[cfg(all(test, feature = "test-util"))] @@ -9655,6 +9655,141 @@ mod tests { assert!(nested.join(STORAGE_FORMAT_FILE).exists()); } + /// The residue a delete leaves once it committed on a disk but before its + /// cleanup pass: a marked data dir and the rollback backup of the same + /// transaction. Returns `(data dir, rollback backup)`. + async fn write_committed_delete_with_rollback_backup( + object_dir: &std::path::Path, + backup_transaction: Uuid, + marker_transaction: Option, + ) -> (std::path::PathBuf, std::path::PathBuf) { + let data_dir = object_dir.join(Uuid::new_v4().to_string()); + fs::create_dir_all(&data_dir).await.expect("data dir should be created"); + fs::write(data_dir.join("part.1"), b"stale") + .await + .expect("stale part should be written"); + if let Some(transaction) = marker_transaction { + fs::write( + data_dir.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)), + [], + ) + .await + .expect("committed delete marker should be written"); + } + let backup = object_dir + .join(backup_transaction.to_string()) + .join(STORAGE_FORMAT_FILE_BACKUP); + fs::create_dir_all(backup.parent().expect("backup parent")) + .await + .expect("rollback dir should be created"); + fs::write(&backup, b"rollback metadata") + .await + .expect("rollback backup should be written"); + (data_dir, backup) + } + + // #6898: the rollback backup of a delete whose committed marker sits in a + // data dir of the same object is residue too, so the empty listing can + // reclaim the whole object directory. + #[tokio::test] + async fn purge_orphan_dir_object_removes_rollback_backup_of_committed_delete() { + let (dir, disk) = make_single_local_disk().await; + let transaction = Uuid::new_v4(); + let object_dir = dir.path().join("bucket/pfx/obj"); + write_committed_delete_with_rollback_backup(&object_dir, transaction, Some(transaction)).await; + + let set = make_set_disks_with(vec![Some(disk)]).await; + let purged = set + .purge_orphan_dir_object("bucket", "pfx/") + .await + .expect("purge should succeed"); + + assert!(purged); + assert!(!dir.path().join("bucket/pfx").exists(), "the whole committed residue tree must go"); + } + + #[tokio::test] + async fn purge_orphan_dir_object_preserves_rollback_backup_of_other_transaction() { + let (dir, disk) = make_single_local_disk().await; + let object_dir = dir.path().join("bucket/pfx/obj"); + let (data_dir, backup) = + write_committed_delete_with_rollback_backup(&object_dir, Uuid::new_v4(), Some(Uuid::new_v4())).await; + + let set = make_set_disks_with(vec![Some(disk)]).await; + set.purge_orphan_dir_object("bucket", "pfx/") + .await + .expect("scan should succeed"); + + assert!(!data_dir.exists(), "the committed data dir is still reclaimed"); + assert!(backup.exists(), "a backup whose own transaction is unproven must remain"); + } + + #[tokio::test] + async fn purge_orphan_dir_object_preserves_rollback_backup_without_marker() { + let (dir, disk) = make_single_local_disk().await; + let object_dir = dir.path().join("bucket/pfx/obj"); + let backup = object_dir.join(Uuid::new_v4().to_string()).join(STORAGE_FORMAT_FILE_BACKUP); + fs::create_dir_all(backup.parent().expect("backup parent")) + .await + .expect("rollback dir should be created"); + fs::write(&backup, b"rollback metadata") + .await + .expect("rollback backup should be written"); + + let set = make_set_disks_with(vec![Some(disk)]).await; + let purged = set + .purge_orphan_dir_object("bucket", "pfx/") + .await + .expect("scan should succeed"); + + assert!(!purged, "a lone rollback backup may still be needed to roll a delete back"); + assert!(backup.exists()); + } + + // Live metadata on one disk blocks the object there and keeps it intact; + // on the other disks only what each disk's own committed marker proves is + // reclaimed, the same rule that already applies to a marked data dir. + #[tokio::test] + async fn purge_orphan_dir_object_keeps_object_with_metadata_on_another_disk() { + let mut dirs = Vec::new(); + let mut disks = Vec::new(); + for _ in 0..4 { + let (dir, disk) = make_single_local_disk().await; + dirs.push(dir); + disks.push(Some(disk)); + } + let transaction = Uuid::new_v4(); + let mut residue = Vec::new(); + for dir in &dirs[..3] { + residue.push( + write_committed_delete_with_rollback_backup(&dir.path().join("bucket/pfx/obj"), transaction, Some(transaction)) + .await, + ); + } + let live_object = dirs[3].path().join("bucket/pfx/obj"); + let live_part = live_object.join(Uuid::new_v4().to_string()).join("part.1"); + fs::create_dir_all(live_part.parent().expect("data dir")) + .await + .expect("live data dir should be created"); + fs::write(&live_part, b"live").await.expect("live part should be written"); + fs::write(live_object.join(STORAGE_FORMAT_FILE), b"meta") + .await + .expect("live metadata should be written"); + + let set = make_set_disks_with(disks).await; + set.purge_orphan_dir_object("bucket", "pfx/") + .await + .expect("scan should succeed"); + + assert!( + live_object.join(STORAGE_FORMAT_FILE).exists() && live_part.exists(), + "the object with metadata on another disk must stay intact there" + ); + for (data_dir, backup) in &residue { + assert!(!data_dir.exists() && !backup.exists(), "this disk's committed residue is reclaimed"); + } + } + // An unreadable directory anywhere under the prefix aborts the purge on // every disk before anything is deleted. #[cfg(unix)] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 0416310f7..7e2879825 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7078,6 +7078,82 @@ async fn pause_delete_object_commit_after_publish(bucket: &str, object: &str) { } } +/// Pauses a delete after its per-disk commit and before the cleanup or rollback +/// pass, the window in which an abandoned delete used to leave metadata-less +/// residue behind. +#[cfg(test)] +struct DeleteCleanupBarrierState { + bucket: String, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(test)] +pub(crate) struct DeleteCleanupBarrier { + state: Arc, +} + +#[cfg(test)] +static DELETE_CLEANUP_BARRIER: std::sync::OnceLock>>> = + std::sync::OnceLock::new(); + +#[cfg(test)] +impl DeleteCleanupBarrier { + pub(crate) fn install(bucket: &str) -> Self { + let state = Arc::new(DeleteCleanupBarrierState { + bucket: bucket.to_string(), + arrived: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let mut slot = DELETE_CLEANUP_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("delete cleanup barrier mutex should not poison"); + assert!(slot.is_none(), "delete cleanup barrier must be unique"); + *slot = Some(Arc::clone(&state)); + Self { state } + } + + pub(crate) async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("delete should reach the cleanup barrier"); + } + + pub(crate) fn release(&self) { + self.state.release.notify_one(); + } +} + +#[cfg(test)] +impl Drop for DeleteCleanupBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = DELETE_CLEANUP_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("delete cleanup barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(test)] +async fn pause_delete_before_cleanup(bucket: &str) { + let barrier = DELETE_CLEANUP_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("delete cleanup barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.bucket == bucket) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + fn persisted_transition_version_with_gate( remote_version: &str, remote_version_state_writer_enabled: bool, @@ -7658,6 +7734,8 @@ impl SetDisks { let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object); let should_rollback = quorum_result.is_err(); + #[cfg(test)] + pause_delete_before_cleanup(bucket).await; let mut rollback_futures = Vec::new(); for (index, err) in errs.iter().enumerate() { // backlog#1158: when rolling back, fan the idempotent undo out to every @@ -8667,6 +8745,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + #[cfg(test)] + pause_delete_before_cleanup(bucket).await; let mut rollback_futures = Vec::new(); let committed_receipt_indices = diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index c7654c1ac..49e02380e 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -14521,15 +14521,19 @@ mod tests { let barrier = TierDeleteChunkTestBarrier::install(stage); let worker_store = store.clone(); let worker_bucket = bucket.clone(); + // Simulate a crash at the barrier. Aborting a caller of the public entry + // point no longer stops the delete, which runs on a detached task, so + // drive the delete below that task. let worker = tokio::spawn(async move { worker_store - .delete_object_with_tier_delete_journal( + .handle_delete_object_with_journal( &worker_bucket, prefix, ObjectOptions { delete_prefix: true, ..Default::default() }, + Some(worker_store.clone()), ) .await }); @@ -14538,6 +14542,11 @@ mod tests { .unwrap_or_else(|_| panic!("chunk delete did not reach crash boundary {stage:?}")); worker.abort(); let _ = worker.await; + assert_eq!( + ctx.detached_mutation_count(), + 0, + "the crash must stop the delete below the detached task, so nothing keeps running after the abort" + ); drop(barrier); let released_bucket_guard = tokio::time::timeout(Duration::from_secs(5), store.acquire_bucket_lifecycle_write_lock(&bucket)) @@ -14683,15 +14692,19 @@ mod tests { let barrier = TierDeleteChunkTestBarrier::install(TierDeleteChunkTestStage::DispatchAuthorized); let worker_store = store.clone(); + // Simulate a crash at the barrier. Aborting a caller of the public entry + // point no longer stops the delete, which runs on a detached task, so + // drive the delete below that task. let worker = tokio::spawn(async move { worker_store - .delete_object_with_tier_delete_journal( + .handle_delete_object_with_journal( bucket, prefix, ObjectOptions { delete_prefix: true, ..Default::default() }, + Some(worker_store.clone()), ) .await }); @@ -14700,6 +14713,11 @@ mod tests { .expect("chunk delete should persist its authorization before corruption injection"); worker.abort(); let _ = worker.await; + assert_eq!( + ctx.detached_mutation_count(), + 0, + "the crash must stop the delete below the detached task, so nothing keeps running after the abort" + ); drop(barrier); let records = store @@ -17536,6 +17554,332 @@ mod tests { .expect("new-generation object must survive the old batch delete"); } + async fn put_detached_delete_fixture(store: &Arc, bucket: &str, object: &str) { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("create detached delete bucket"); + // Large enough for a data dir, so the delete commit leaves a marked + // data dir and a rollback backup until its cleanup pass runs. + let mut reader = PutObjReader::from_vec(vec![7u8; 2 * 1024 * 1024]); + store + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("put detached delete fixture"); + } + + async fn assert_delete_lock_moved_to_detached_task(store: &Arc, bucket: &str, object: &str) { + let lock = store + .handle_new_ns_lock(bucket, object) + .await + .expect("namespace lock handle for the deleted key"); + lock.get_write_lock_quiet(Duration::from_millis(200)) + .await + .expect_err("the detached delete must keep its object lock after the caller is dropped"); + } + + fn assert_no_delete_residue(temp_dir: &std::path::Path, bucket: &str, object: &str, disks: usize) { + for disk in 0..disks { + let object_dir = temp_dir.join(format!("pool0/set0/disk{disk}")).join(bucket).join(object); + assert!( + !object_dir.exists(), + "disk {disk} must hold neither metadata, rollback backup nor data of the deleted object: {object_dir:?}" + ); + } + } + + // #6898: a DeleteObjects request dropped between the per-disk commit and the + // cleanup pass (client disconnect) must not leave metadata-less residue. + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn delete_objects_finishes_cleanup_after_caller_is_dropped() { + let temp = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "detached-batch-delete", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("detached-batch-delete-{}", uuid::Uuid::new_v4()); + let object = "residue/batch.bin"; + put_detached_delete_fixture(&store, &bucket, object).await; + + let barrier = crate::set_disk::DeleteCleanupBarrier::install(&bucket); + let delete = store.delete_objects_with_tier_delete_journal_and_accounting( + &bucket, + vec![ObjectToDelete { + object_name: object.to_string(), + ..Default::default() + }], + ObjectOptions::default(), + ); + tokio::select! { + () = barrier.wait_until_paused() => {} + _ = delete => panic!("batch delete must pause between its commit and its cleanup"), + } + assert_eq!(ctx.detached_mutation_count(), 1, "the dropped request must leave its delete running"); + assert_delete_lock_moved_to_detached_task(&store, &bucket, object).await; + + barrier.release(); + assert!( + ctx.wait_for_detached_mutations(Duration::from_secs(30)).await, + "the detached batch delete should finish" + ); + assert_no_delete_residue(temp.path(), &bucket, object, 4); + store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect_err("the object must be deleted"); + } + + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn delete_object_finishes_cleanup_after_caller_is_dropped() { + let temp = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "detached-single-delete", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("detached-single-delete-{}", uuid::Uuid::new_v4()); + let object = "residue/single.bin"; + put_detached_delete_fixture(&store, &bucket, object).await; + + let barrier = crate::set_disk::DeleteCleanupBarrier::install(&bucket); + let delete = store.delete_object_with_tier_delete_journal(&bucket, object, ObjectOptions::default()); + tokio::select! { + () = barrier.wait_until_paused() => {} + _ = delete => panic!("single delete must pause between its commit and its cleanup"), + } + assert_eq!(ctx.detached_mutation_count(), 1, "the dropped request must leave its delete running"); + assert_delete_lock_moved_to_detached_task(&store, &bucket, object).await; + + barrier.release(); + assert!( + ctx.wait_for_detached_mutations(Duration::from_secs(30)).await, + "the detached single delete should finish" + ); + assert_no_delete_residue(temp.path(), &bucket, object, 4); + store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect_err("the object must be deleted"); + } + + // A recursive (force) delete runs under its caller's bucket lifecycle write + // lock and carries only that lock's fence. If the caller is dropped, the + // detached delete must still hold the lock: otherwise a PUT under the prefix + // succeeds and is then removed by the delete that is still running. + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn recursive_delete_keeps_bucket_lock_after_caller_is_dropped() { + let temp = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "detached-recursive-delete", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("detached-recursive-delete-{}", uuid::Uuid::new_v4()); + let old_object = "folder/old.bin"; + let new_object = "folder/new.bin"; + put_detached_delete_fixture(&store, &bucket, old_object).await; + + // The S3 force-delete path: the caller owns the lock, opts carry its fence. + let guard = store + .lock_bucket_for_recursive_delete(&bucket) + .await + .expect("recursive delete should lock the bucket"); + let mut opts = ObjectOptions { + delete_prefix: true, + ..Default::default() + }; + opts.add_bucket_lifecycle_lock_guard(&guard); + let barrier = crate::store::object::DeleteAfterObjectLockSnapshotBarrier::install(&bucket); + let delete = store.delete_object_with_tier_delete_journal_and_guards(&bucket, "folder/", opts, Some(guard)); + tokio::select! { + () = barrier.wait_until_paused() => {} + _ = delete => panic!("recursive delete must pause inside its detached task"), + } + assert_eq!(ctx.detached_mutation_count(), 1, "the dropped caller must leave its delete running"); + + let put_store = Arc::clone(&store); + let put_bucket = bucket.clone(); + let mut put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(b"written while the delete runs".to_vec()); + put_store + .put_object(&put_bucket, new_object, &mut reader, &ObjectOptions::default()) + .await + }); + assert!( + tokio::time::timeout(Duration::from_millis(500), &mut put).await.is_err(), + "a PUT under the prefix must wait while the detached recursive delete runs" + ); + + barrier.release(); + assert!( + ctx.wait_for_detached_mutations(Duration::from_secs(30)).await, + "the detached recursive delete should finish" + ); + put.await + .expect("PUT task should join") + .expect("the PUT should succeed once the recursive delete has finished"); + store + .get_object_info(&bucket, old_object, &ObjectOptions::default()) + .await + .expect_err("the recursive delete must remove the object it covered"); + store + .get_object_info(&bucket, new_object, &ObjectOptions::default()) + .await + .expect("the PUT that waited for the recursive delete must survive it"); + for disk in 0..4 { + let metadata = temp + .path() + .join(format!("pool0/set0/disk{disk}")) + .join(&bucket) + .join(new_object) + .join(crate::disk::STORAGE_FORMAT_FILE); + assert!( + metadata.exists(), + "disk {disk} must keep the object written after the delete: {metadata:?}" + ); + } + } + + // Lifecycle expiry holds the table-bucket publication read lock and passes + // only its lock-lost signal to the delete. A cancelled expiry worker must not + // release that lock while its detached delete is still running. + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn lifecycle_expiry_keeps_publication_lock_after_worker_is_dropped() { + let temp = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "detached-lifecycle-expiry", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("detached-lifecycle-expiry-{}", uuid::Uuid::new_v4()); + let object = "expired.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create lifecycle bucket"); + let mut reader = PutObjReader::from_vec(b"expired data".to_vec()); + let object_info = store + .put_object(&bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("put expired object"); + let bucket_incarnation_id = store.bucket_incarnation_id(&bucket).await.expect("read bucket incarnation"); + + let barrier = crate::store::object::DeleteAfterObjectLockSnapshotBarrier::install(&bucket); + let expiry_store = Arc::clone(&store); + let expiry = tokio::spawn(async move { + crate::bucket::lifecycle::bucket_lifecycle_ops::apply_expiry_on_non_transitioned_objects( + expiry_store, + &object_info, + &crate::bucket::lifecycle::lifecycle::Event { + action: crate::bucket::lifecycle::lifecycle::IlmAction::DeleteAction, + ..Default::default() + }, + &crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Scanner, + bucket_incarnation_id, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("lifecycle delete should pause inside its detached task"); + // A cancelled worker drops its future while the delete keeps running. + expiry.abort(); + assert!( + expiry + .await + .expect_err("aborted expiry worker should not join") + .is_cancelled(), + "expiry worker should be cancelled" + ); + assert_eq!(ctx.detached_mutation_count(), 1, "the cancelled worker must leave its delete running"); + + let publication_lock = store + .handle_new_ns_lock(&bucket, rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH) + .await + .expect("table-bucket publication lock handle"); + publication_lock + .get_write_lock_quiet(Duration::from_millis(300)) + .await + .expect_err("table-bucket publication must wait for the detached lifecycle delete"); + + barrier.release(); + assert!( + ctx.wait_for_detached_mutations(Duration::from_secs(30)).await, + "the detached lifecycle delete should finish" + ); + let publication_guard = publication_lock + .get_write_lock_quiet(crate::set_disk::get_lock_acquire_timeout()) + .await + .expect("table-bucket publication should proceed once the lifecycle delete has finished"); + drop(publication_guard); + store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect_err("the detached lifecycle delete should still remove the expired object"); + } + + // #6898: residue that an older build stranded (every disk committed the + // delete, the cleanup pass never ran) is reclaimed by deleting the key again. + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn repeated_delete_objects_finishes_committed_delete_residue() { + use crate::disk::DiskAPI as _; + + for drives in [1, 4] { + let temp = tempfile::tempdir().expect("create temp store dir"); + let (_ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "residue-retry", &[drives])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("residue-retry-{drives}-{}", uuid::Uuid::new_v4()); + let object = "residue/retry.bin"; + put_detached_delete_fixture(&store, &bucket, object).await; + + let transaction = uuid::Uuid::new_v4(); + let disks = store.pools[0].disk_set[0].disks.read().await.clone(); + for disk in disks.iter().flatten() { + let errors = disk + .delete_versions( + &bucket, + vec![crate::disk::FileInfoVersions { + name: object.to_string(), + versions: vec![rustfs_filemeta::FileInfo { + name: object.to_string(), + ..Default::default() + }], + ..Default::default() + }], + crate::disk::DeleteOptions { + old_data_dir: Some(transaction), + ..Default::default() + }, + ) + .await; + assert!(errors.iter().all(Option::is_none), "the delete should commit on every disk: {errors:?}"); + } + for disk in 0..drives { + let object_dir = temp.path().join(format!("pool0/set0/disk{disk}")).join(&bucket).join(object); + assert!(!object_dir.join(crate::disk::STORAGE_FORMAT_FILE).exists()); + assert!( + object_dir + .join(transaction.to_string()) + .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP) + .exists(), + "disk {disk} should hold the stranded rollback backup" + ); + } + + let (_deleted, errors) = store + .delete_objects( + &bucket, + vec![ObjectToDelete { + object_name: object.to_string(), + ..Default::default() + }], + ObjectOptions::default(), + ) + .await; + assert!(errors.iter().all(Option::is_none), "a missing key deletes successfully: {errors:?}"); + assert_no_delete_residue(temp.path(), &bucket, object, drives); + } + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn prefix_delete_serializes_with_concurrent_object_lock_metadata_updates() { diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index ec7a3aac4..038e436e0 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -9644,6 +9644,67 @@ mod test { } } + // #6898: what an abandoned DeleteObjects left on every disk (rollback + // backup plus marked data dir) no longer blocks DeleteBucket after an empty + // recursive listing of the bucket. + #[tokio::test] + async fn empty_recursive_listing_purges_stranded_delete_rollback_backups() { + use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks}; + use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions}; + + let (dirs, store) = isolated_store_over_temp_disks().await; + let bucket = "listing-purge-rollback-backup"; + init_bucket_metadata_sys(store.clone(), Vec::new()).await; + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created with authoritative metadata"); + let transaction = uuid::Uuid::new_v4(); + for object in ["repro/obj-00001.bin", "repro/obj-00002.bin"] { + let data_dir = uuid::Uuid::new_v4(); + for dir in &dirs { + let object_dir = dir.path().join(bucket).join(object); + let residue = object_dir.join(data_dir.to_string()); + tokio::fs::create_dir_all(&residue) + .await + .expect("committed data dir should be created"); + tokio::fs::write(residue.join("part.1"), b"stale") + .await + .expect("stale part should be written"); + tokio::fs::write( + residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)), + [], + ) + .await + .expect("committed delete marker should be written"); + let backup = object_dir.join(transaction.to_string()); + tokio::fs::create_dir_all(&backup) + .await + .expect("rollback dir should be created"); + tokio::fs::write(backup.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP), b"rollback metadata") + .await + .expect("rollback backup should be written"); + } + } + + let result = store + .clone() + .list_objects_generic(bucket, "", None, None, 1000, false) + .await + .expect("recursive bucket listing should succeed"); + assert!(result.objects.is_empty()); + for dir in &dirs { + assert!( + !dir.path().join(bucket).join("repro").exists(), + "the empty listing should reclaim the stranded delete residue" + ); + } + store + .delete_bucket(bucket, &DeleteBucketOptions::default()) + .await + .expect("the bucket is empty once the residue is reclaimed"); + } + #[tokio::test] async fn empty_recursive_bucket_listing_purges_orphan_directory_prefixes() { use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks}; diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 444d00ab8..988da280f 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -2456,6 +2456,27 @@ fn return_batch_delete_lock_error_with_accounting( (deleted, errors, vec![None; objects.len()]) } +const EVENT_DETACHED_DELETE_FAILED: &str = "detached_delete_failed"; +const LOG_SUBSYSTEM_STORE_DELETE: &str = "store_delete"; + +/// A detached delete only fails to join when it panicked or the runtime is +/// shutting down; report either as a failed delete to the waiting caller. +fn detached_delete_failed(operation: &'static str, err: tokio::task::JoinError) -> Error { + error!( + event = EVENT_DETACHED_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_DELETE, + operation, + reason = if err.is_panic() { "task_panicked" } else { "task_cancelled" }, + "Detached delete task failed" + ); + if err.is_cancelled() { + Error::OperationCanceled + } else { + Error::Unexpected + } +} + fn sorted_unique_delete_object_names(objects: &[ObjectToDelete]) -> Vec<&str> { let mut object_names: Vec<&str> = objects.iter().map(|object| object.object_name.as_str()).collect(); object_names.sort_unstable(); @@ -4798,19 +4819,54 @@ impl ECStore { purged } + /// Delete one object (or a prefix) on behalf of a caller that may be + /// dropped mid-way, such as an S3 request whose client disconnects. + /// + /// The delete runs to completion on a detached task that owns its locks + /// and both on-disk phases, so the object is left either intact or fully + /// deleted, never with its metadata gone and its data still in place. pub async fn delete_object_with_tier_delete_journal( self: &Arc, bucket: &str, object: &str, opts: ObjectOptions, ) -> Result { - let result = self - .handle_delete_object_with_journal(bucket, object, opts, Some(Arc::clone(self))) - .await; - if result.is_ok() { - list_objects::observe_list_objects_mutation(self, bucket).await; - } - result + self.delete_object_with_tier_delete_journal_and_guards(bucket, object, opts, None) + .await + } + + /// [`Self::delete_object_with_tier_delete_journal`] for a caller that + /// holds namespace locks covering the delete and passes only their fences + /// in `opts`, such as the bucket lifecycle write lock of a recursive delete. + /// + /// The guards move into the detached task and are released only after + /// the delete has finished. Dropping the caller therefore cannot let a + /// writer into the scope they exclude while the delete is still running. + pub async fn delete_object_with_tier_delete_journal_and_guards( + self: &Arc, + bucket: &str, + object: &str, + opts: ObjectOptions, + guards: impl IntoIterator, + ) -> Result { + let guards: Vec<_> = guards.into_iter().collect(); + let store = Arc::clone(self); + let bucket = bucket.to_owned(); + let object = object.to_owned(); + self.ctx + .run_detached_mutation(async move { + let result = store + .handle_delete_object_with_journal(&bucket, &object, opts, Some(Arc::clone(&store))) + .await; + if result.is_ok() { + list_objects::observe_list_objects_mutation(&store, &bucket).await; + } + // Keep: the only use of `guards` here, it moves them into the task so the locks outlive a dropped caller. + drop(guards); + result + }) + .await + .unwrap_or_else(|err| Err(detached_delete_failed("delete_object", err))) } pub async fn delete_objects_with_tier_delete_journal( @@ -4819,30 +4875,43 @@ impl ECStore { objects: Vec, opts: ObjectOptions, ) -> (Vec, Vec>) { - let result = self - .handle_delete_objects_with_journal(bucket, objects, opts, Some(Arc::clone(self))) + let (deleted, errors, _) = self + .delete_objects_with_tier_delete_journal_and_accounting(bucket, objects, opts) .await; - let success_count = result.1.iter().filter(|err| err.is_none()).count(); - if success_count > 0 { - list_objects::observe_list_objects_mutations(self, bucket, success_count).await; - } - result + (deleted, errors) } + /// Batch form of [`Self::delete_object_with_tier_delete_journal`]: the + /// whole batch, including its object locks, runs on one detached task. pub async fn delete_objects_with_tier_delete_journal_and_accounting( self: &Arc, bucket: &str, objects: Vec, opts: ObjectOptions, ) -> (Vec, Vec>, Vec>) { - let result = self - .handle_delete_objects_with_journal_and_accounting(bucket, objects, opts, Some(Arc::clone(self))) - .await; - let success_count = result.1.iter().filter(|err| err.is_none()).count(); - if success_count > 0 { - list_objects::observe_list_objects_mutations(self, bucket, success_count).await; - } - result + let store = Arc::clone(self); + let bucket = bucket.to_owned(); + let object_count = objects.len(); + self.ctx + .run_detached_mutation(async move { + let result = store + .handle_delete_objects_with_journal_and_accounting(&bucket, objects, opts, Some(Arc::clone(&store))) + .await; + let success_count = result.1.iter().filter(|err| err.is_none()).count(); + if success_count > 0 { + list_objects::observe_list_objects_mutations(&store, &bucket, success_count).await; + } + result + }) + .await + .unwrap_or_else(|err| { + let err = detached_delete_failed("delete_objects", err); + ( + vec![DeletedObject::default(); object_count], + vec![Some(err); object_count], + vec![None; object_count], + ) + }) } #[instrument(skip(self))] diff --git a/docs/architecture/global-state-crate-split-plan.md b/docs/architecture/global-state-crate-split-plan.md index 59bca1ae0..6d50659fc 100644 --- a/docs/architecture/global-state-crate-split-plan.md +++ b/docs/architecture/global-state-crate-split-plan.md @@ -33,7 +33,7 @@ The guard pins the production files allowed to reference `rustfs_ecstore::api::g - `rustfs/src/storage/storage_api.rs` -That boundary keeps only bootstrap writes and lifecycle controls (`set_global_endpoints`, `set_global_region`, `set_global_rustfs_port`, `set_object_store_resolver`, `shutdown_background_services`, `update_erasure_type`). Read-only runtime getters are exported through `rustfs_ecstore::api::runtime` and consumed through the local storage facade. A new direct use either moves behind an existing owner-local boundary or updates this plan and the guard in the same reviewed change. +That boundary keeps only bootstrap writes and lifecycle controls (`set_global_endpoints`, `set_global_region`, `set_global_rustfs_port`, `set_object_store_resolver`, `shutdown_background_services`, `update_erasure_type`, `wait_for_detached_mutations`). Read-only runtime getters are exported through `rustfs_ecstore::api::runtime` and consumed through the local storage facade. A new direct use either moves behind an existing owner-local boundary or updates this plan and the guard in the same reviewed change. ## Fallback Removal Plan diff --git a/docs/architecture/global-state-inventory.md b/docs/architecture/global-state-inventory.md index 345d7444b..e46375e15 100644 --- a/docs/architecture/global-state-inventory.md +++ b/docs/architecture/global-state-inventory.md @@ -27,7 +27,7 @@ Runtime ownership handles that exist today. Reads go through `rustfs_ecstore::ap | `global_tier_config_mgr` | Tier config manager | Reads and reloads stay behind this helper. | | `bucket_monitor` | Replication bandwidth monitor | Replication pool/stat handles are projected into RustFS wrapper types at the storage boundary. | | `global_lock_client`, `global_lock_clients` | `GLOBAL_LOCAL_LOCK_CLIENT`, `GLOBAL_LOCK_CLIENTS` (`crates/ecstore/src/runtime/global.rs`) | Preserve lock quorum and client selection; the process-level `GLOBAL_LOCK_MANAGER` stays separate. | -| `boot_time`, `deployment_id`, `region`, `rustfs_port` | `GLOBAL_BOOT_TIME`, deployment id, region, and port state (`crates/ecstore/src/runtime/global.rs`) | Scalar writes remain behind the `api::global` setters (`set_global_endpoints`, `set_global_region`, `set_global_rustfs_port`, `set_object_store_resolver`, `shutdown_background_services`, `update_erasure_type`). | +| `boot_time`, `deployment_id`, `region`, `rustfs_port` | `GLOBAL_BOOT_TIME`, deployment id, region, and port state (`crates/ecstore/src/runtime/global.rs`) | Scalar writes remain behind the `api::global` setters (`set_global_endpoints`, `set_global_region`, `set_global_rustfs_port`, `set_object_store_resolver`, `shutdown_background_services`, `update_erasure_type`, `wait_for_detached_mutations`). | Owner-helper handles outside the runtime-source list stay inside their owner and are reached through owner functions: `GLOBAL_EVENT_NOTIFIER` (`crates/ecstore/src/runtime/global.rs`); `GLOBAL_NOTIFICATION_SYS`, `EVENT_DISPATCH_HOOK`, `GLOBAL_PROCESSORS`, `INTERNODE_DATA_TRANSPORT`, `GLOBAL_BUCKET_TARGET_SYS`, `GLOBAL_CONFIG_SYS`, `GLOBAL_STORAGE_CLASS`, `WORKLOAD_ADMISSION_SNAPSHOT_PROVIDER` (ECStore owner modules); `GLOBAL_SERVER_CONFIG` (`crates/config/src/server_config.rs`); `GLOBAL_HEAL_RUNTIME`, `GLOBAL_AHM_SERVICES_CANCEL_TOKEN` (`crates/heal/src/lib.rs`); `GLOBAL_KMS_SERVICE_MANAGER` (`crates/kms/src/service_manager.rs`); `GLOBAL_CAPACITY_MANAGER` (`crates/object-capacity/src/capacity_manager.rs`); `APP_CONTEXT_SINGLETON` (`rustfs/src/app/context/global.rs`). diff --git a/rustfs/src/app/object/delete.rs b/rustfs/src/app/object/delete.rs index 351d78b7c..0457f6521 100644 --- a/rustfs/src/app/object/delete.rs +++ b/rustfs/src/app/object/delete.rs @@ -976,6 +976,8 @@ impl DefaultObjectUsecase { // Lock order is bucket lifecycle, then object/commit locks in storage. // Keep this guard alive through the physical delete: a preflight without // writer exclusion could authorize one subtree and delete a newer one. + // The delete takes ownership of it and releases it only when it has + // finished, even if this request is dropped first. let recursive_delete_guard = if force_header { Some( store @@ -1076,7 +1078,7 @@ impl DefaultObjectUsecase { let obj_info = { match store - .delete_object_with_tier_delete_journal(&bucket, &key, opts.clone()) + .delete_object_with_tier_delete_journal_and_guards(&bucket, &key, opts.clone(), recursive_delete_guard) .await { Ok(obj) => obj, @@ -1118,7 +1120,6 @@ impl DefaultObjectUsecase { } } }; - drop(recursive_delete_guard); if force_delete { let _ = invalidate_object_data_cache_prefix_after_delete(&cache_adapter, &bucket, &key).await; diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index 50f855662..3234eca22 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/rustfs/src/startup_shutdown.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::storage_api::startup::shutdown::shutdown_background_services; +use crate::storage_api::startup::shutdown::{shutdown_background_services, wait_for_detached_mutations}; use crate::{ server::{ SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, @@ -50,6 +50,7 @@ const BACKGROUND_SERVICE_CAPACITY: &str = "capacity"; const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown"; const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown"; const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state"; +const DETACHED_MUTATION_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); fn join_failure_reason(error: &tokio::task::JoinError) -> &'static str { if error.is_cancelled() { @@ -199,6 +200,46 @@ fn background_shutdown_steps(enable_scanner: bool, enable_heal: bool) -> Vec( + heal_handoff_complete: bool, + detached_mutation_timeout: std::time::Duration, + wait_for_detached_mutations: W, + clear_unclean_shutdown_markers: ClearFut, +) -> bool +where + W: FnOnce(std::time::Duration) -> WaitFut, + WaitFut: Future, + ClearFut: Future, +{ + let detached_mutations_drained = wait_for_detached_mutations(detached_mutation_timeout).await; + if !detached_mutations_drained { + warn!( + target: "rustfs::main::handle_shutdown", + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "detached_mutations", + state = "stop_failed", + reason = "timeout", + timeout_secs = detached_mutation_timeout.as_secs(), + "Detached mutations did not finish before shutdown; retaining unclean-shutdown markers" + ); + } + if !(heal_handoff_complete && detached_mutations_drained) { + return false; + } + clear_unclean_shutdown_markers.await; + true +} + pub(crate) async fn run_startup_shutdown_sequence( state_manager: &ServiceStateManager, shutdown_signal: ShutdownSignal, @@ -422,11 +463,15 @@ pub(crate) async fn run_startup_shutdown_sequence( console_shutdown_handle.shutdown().await; } shutdown_optional_runtime_services(optional_runtime_shutdowns).await; - // The data plane is drained: record this shutdown as clean so the next - // startup skips the unclean-restart erasure-set heal. - if heal_handoff_complete { - rustfs_heal::heal::clear_unclean_shutdown_markers().await; - } + // FTP, SFTP and WebDAV can start deletes too: wait for detached mutations + // only after every front end has stopped. + finish_detached_mutations_and_clear_unclean_shutdown_markers( + heal_handoff_complete, + DETACHED_MUTATION_SHUTDOWN_TIMEOUT, + wait_for_detached_mutations, + rustfs_heal::heal::clear_unclean_shutdown_markers(), + ) + .await; state_manager.update(ServiceState::Stopped); info!( target: "rustfs::main::handle_shutdown", @@ -605,7 +650,8 @@ pub(crate) async fn run_embedded_server_shutdown( #[cfg(test)] mod tests { use super::{ - BackgroundShutdownStep, EmbeddedRuntimeOwners, background_shutdown_steps, finish_embedded_server_cleanup, + BackgroundShutdownStep, EmbeddedRuntimeOwners, background_shutdown_steps, + finish_detached_mutations_and_clear_unclean_shutdown_markers, finish_embedded_server_cleanup, run_embedded_server_drop_cleanup, signal_embedded_startup_shutdown, }; use crate::server::ShutdownHandle; @@ -628,6 +674,61 @@ mod tests { assert!(background_shutdown_steps(false, false).is_empty()); } + async fn finish_shutdown_with_marker( + marker: &std::path::Path, + heal_handoff_complete: bool, + timeout: Duration, + detached_mutation: &mut tokio::task::JoinHandle<()>, + ) -> bool { + finish_detached_mutations_and_clear_unclean_shutdown_markers( + heal_handoff_complete, + timeout, + |timeout| async move { tokio::time::timeout(timeout, detached_mutation).await.is_ok() }, + async { + std::fs::remove_file(marker).expect("unclean-shutdown marker should be removable"); + }, + ) + .await + } + + #[tokio::test] + async fn unclean_shutdown_markers_are_kept_while_a_detached_mutation_runs() { + let temp_dir = tempfile::tempdir().expect("temp dir should create"); + let marker = temp_dir.path().join("unclean-shutdown"); + std::fs::write(&marker, b"").expect("unclean-shutdown marker should be written"); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); + let mut detached_mutation = tokio::spawn(async move { + let _ = release_rx.await; + }); + + assert!( + !finish_shutdown_with_marker(&marker, true, Duration::from_millis(50), &mut detached_mutation).await, + "a detached mutation still running at the deadline must keep the markers" + ); + assert!(marker.exists(), "the next startup must still see the unclean-shutdown marker"); + + release_tx.send(()).expect("detached mutation should be released"); + assert!( + finish_shutdown_with_marker(&marker, true, Duration::from_secs(5), &mut detached_mutation).await, + "the markers should be cleared once every detached mutation has finished" + ); + assert!(!marker.exists(), "a clean shutdown should clear the unclean-shutdown marker"); + } + + #[tokio::test] + async fn unclean_shutdown_markers_are_kept_after_a_failed_heal_handoff() { + let temp_dir = tempfile::tempdir().expect("temp dir should create"); + let marker = temp_dir.path().join("unclean-shutdown"); + std::fs::write(&marker, b"").expect("unclean-shutdown marker should be written"); + let mut detached_mutation = tokio::spawn(async {}); + + assert!( + !finish_shutdown_with_marker(&marker, false, Duration::from_secs(5), &mut detached_mutation).await, + "a failed heal handoff must keep the markers" + ); + assert!(marker.exists(), "the next startup must still see the unclean-shutdown marker"); + } + #[tokio::test] async fn signal_embedded_startup_shutdown_signals_handle_and_cancels_token() { let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 91d76c57d..142948502 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -504,7 +504,7 @@ pub(crate) mod ecstore_event { pub(crate) mod ecstore_global { pub(crate) use rustfs_ecstore::api::global::{ - set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, + set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, wait_for_detached_mutations, }; } @@ -1342,6 +1342,10 @@ pub(crate) fn shutdown_background_services() { ecstore_global::shutdown_background_services(); } +pub(crate) async fn wait_for_detached_mutations(timeout: std::time::Duration) -> bool { + ecstore_global::wait_for_detached_mutations(timeout).await +} + pub(crate) fn shutdown_background_monitors() { rustfs_ecstore::shutdown_background_monitors(); } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 69cdfd774..b7528a860 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -423,7 +423,7 @@ pub(crate) mod startup { pub(crate) mod shutdown { pub(crate) use crate::storage::storage_api::{ mark_get_metadata_read_version_coalescing_service_ready, shutdown_background_monitors, shutdown_background_services, - store_compression_total_in_backend, + store_compression_total_in_backend, wait_for_detached_mutations, }; }