From 99ae115731f2e63c2f429bf23d5a5cc6586432c1 Mon Sep 17 00:00:00 2001 From: cppudge Date: Thu, 8 Oct 2026 16:58:14 +0700 Subject: [PATCH] fix(ecstore): make deletes cancellation-safe and reclaim their residue (#8378) * fix(ecstore): keep object deletes running when the caller is dropped DeleteObject and DeleteObjects commit on every disk and then run a separate cleanup pass, both inside the request future. When the client disconnects, hyper drops that future and the delete stops between the two passes: xl.meta is gone, but the rollback backup and the marked data dir stay on disk, hidden from listings and never reclaimed (#6898). Run the ECStore delete entry points on a detached task tracked by the instance context. The task owns the object locks, both disk passes and any rollback, so dropping the caller only stops the wait. Server shutdown waits up to 30s for detached deletes after the HTTP drain. Two smaller leaks of the same kind: a cancelled data-dir delete no longer leaves its snapshot-lease entry marked as deleting forever, and DiskStore::delete_versions releases its waiting slot through a guard. Signed-off-by: Petr Slavyagin * fix(ecstore): finish committed delete residue on repeated deletes Residue that an older build stranded (#6898) has no xl.meta, so deleting the key again reports it missing and leaves everything in place. On a disk, a delete-data. marker in one of the object's data dirs proves that transaction T removed the metadata there and committed: the data it marked can no longer be read, healed or rolled back on that disk. When delete_versions or delete_version finds no xl.meta, remove what the cleanup pass of every transaction proven that way would have removed, under the object's metadata lease. The rollback backup goes before the marked data dirs, so an attempt cut short still leaves the marker for the next one. Reserved markers, rollback backups without a committed marker and directories that also hold xl.meta or other keys are left alone. The disk result stays FileNotFound; the extra cost is one directory read per missing key inside the disk call that already runs, with no new RPC, and the read stops at the first child that is not a data dir. Finished transactions are counted in rustfs_disk_delete_residue_finished_total instead of logged per object. A single DeleteObject in an unversioned bucket answers a missing key from the metadata read quorum without reaching the disks, so existing residue is reclaimed by DeleteObjects or by the empty-listing purge. Signed-off-by: Petr Slavyagin * fix(ecstore): purge the rollback backup of a committed delete on listing After an empty listing, the orphan purge (#7955, #8261) already removes data dirs that carry a committed delete-data. marker, but it treated the //xl.meta.bkp rollback backup of the same delete as data and blocked the whole object directory and its ancestors. Residue left by an abandoned delete (#6898) therefore survived the listing, and DeleteBucket kept failing with BucketNotEmpty. Treat a UUID dir that holds only xl.meta.bkp as purgeable when a purgeable data dir of the same object on the same disk carries the committed marker of that transaction. A backup without such a marker still blocks as before, since it may be needed to roll the delete back. The other purge gates are unchanged: every disk of the set must be online, and the backoff applies. xl.meta or unmarked data on any disk blocks that directory and its ancestors on every disk. Like a marked data dir before, a backup below such a directory is still removed from a disk that holds no xl.meta there and proves the transaction with its own marker. Signed-off-by: Petr Slavyagin * fix(ecstore): move caller-held delete locks into the detached delete A force (recursive) delete takes the bucket lifecycle write lock in the S3 handler and passes only its fence down, so the store does not take that lock again. Since the delete runs on a detached task, dropping the request released the lock while the delete kept running: a PUT under the same prefix could succeed and then be removed by the old delete. Lifecycle expiry had the same shape with its table-bucket publication read lock, of which only the lock-lost signal reaches the delete. Add ECStore::delete_object_with_tier_delete_journal_and_guards. It moves the caller's guards into the detached task and releases them after the delete and its listing bookkeeping. The force-delete handler and lifecycle expiry hand their guards over; DeleteObjects holds none. Signed-off-by: Petr Slavyagin * fix(startup): keep unclean-shutdown markers while deletes still run Shutdown waits up to 30s for detached deletes, but it cleared the unclean-shutdown markers even when that wait timed out. A delete then cut short by the runtime teardown went unnoticed: the next startup skipped the erasure-set heal that the marker triggers. Clear the markers only when the heal handoff completed and every detached mutation finished, and log that they are kept otherwise. Wait after the FTP, FTPS, SFTP and WebDAV servers have stopped as well, since their deletes take the same detached path. Signed-off-by: Petr Slavyagin * test(ecstore): simulate tier-delete crashes below the detached delete The chunk crash-boundary tests model a crash by aborting the task that runs delete_object_with_tier_delete_journal while it is paused at a TierDeleteChunkTestBarrier. That entry point now runs the delete on a detached task, so aborting the caller no longer stops it: the delete resumed on the old store once the barrier was dropped, and building the restarted store on the same disks tripped the tier-delete journal proof assertion (tier_delete_chunk_crash_boundaries_resume_without_skipping_sources failed in CI). Run the delete through handle_delete_object_with_journal, the body below the detached task, so the abort stops it at the barrier as before. tier_delete_chunk_missing_child_with_journals_fails_closed used the same pattern; its resumed delete only failed fast because the tier rejects removes, so it gets the same change. Signed-off-by: Petr Slavyagin --------- Signed-off-by: Petr Slavyagin --- crates/ecstore/src/api/mod.rs | 2 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 4 +- crates/ecstore/src/disk/disk_store.rs | 38 +- crates/ecstore/src/disk/local.rs | 690 ++++++++++++++++-- crates/ecstore/src/runtime/global.rs | 7 + crates/ecstore/src/runtime/instance.rs | 59 ++ .../src/set_disk/core/io_primitives.rs | 59 +- crates/ecstore/src/set_disk/mod.rs | 139 +++- crates/ecstore/src/set_disk/ops/object.rs | 80 ++ crates/ecstore/src/store/init.rs | 348 ++++++++- crates/ecstore/src/store/list_objects.rs | 61 ++ crates/ecstore/src/store/object.rs | 113 ++- .../global-state-crate-split-plan.md | 2 +- docs/architecture/global-state-inventory.md | 2 +- rustfs/src/app/object/delete.rs | 5 +- rustfs/src/startup_shutdown.rs | 115 ++- rustfs/src/storage/storage_api.rs | 6 +- rustfs/src/storage_api.rs | 2 +- 18 files changed, 1620 insertions(+), 112 deletions(-) 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, }; }