fix(ecstore): make delete_volume non-recursive by default to prevent bucket-heal wipe (backlog#799 B1) (#4339)

* fix(ecstore): make delete_volume non-recursive by default to prevent bucket-heal wipe (backlog#799 B1)

`delete_volume` unconditionally `remove_dir_all`'d the whole bucket tree, and the
bucket-heal "remove" branch called it fire-and-forget on every local disk. A
mis-classified "dangling" bucket (or a non-force S3 DeleteBucket on a populated
bucket) was therefore recursively wiped — a potential whole-bucket data loss.
The `VolumeNotEmpty` -> recreate/`BucketNotEmpty` handling already present in
both delete_bucket paths was dead code because the primitive never refused.

Add an explicit `force_delete` flag to `DiskAPI::delete_volume` and default the
non-force path to a non-recursive `remove_dir` (rmdir), which fails atomically
with `VolumeNotEmpty` if the bucket still holds any object data. Only an explicit
force delete (S3 force bucket delete) removes recursively. Mirrors MinIO's
`xlStorage.DeleteVol` (`Remove` vs `RemoveAll`).

- Trait + all impls (local behavior, dispatch, disk_store, remote RPC) take the
  flag; the gRPC `DeleteVolumeRequest` gains a `force` field (proto3 default
  false → old peers get the safe non-recursive behavior on rolling upgrade).
- Heal remove branch passes `false` and no longer discards the result: a
  `VolumeNotEmpty` refusal is logged (the bucket is not dangling) instead of
  wiping data.
- Both `delete_bucket` paths pass `opts.force`, activating the previously-dead
  `VolumeNotEmpty` -> `BucketNotEmpty`/recreate handling (correct S3 semantics).

Adds a regression test: non-force delete of a non-empty bucket returns
VolumeNotEmpty and preserves the data; force delete removes it.

Design converged by two independent expert reviews (MinIO-fidelity +
defense-in-depth) referencing MinIO xl-storage.go. Refs backlog#799 (B1),
issue rustfs/backlog#850. The safety expert's deeper hardening (typed
capability instead of a bool, trash-instead-of-in-place for force, quorum
re-verification of dangling) is noted on #850 as follow-up.

* fix(ecstore): reword 'mis-classified' -> 'misclassified' to satisfy typos (backlog#799 B1)

* fix(rustfs): thread force_delete through StorageDiskRpcExt::delete_volume + test literal (backlog#799 B1)

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -553,7 +553,7 @@ impl PeerS3Client for LocalPeerS3Client {
             .ok_or(Error::VolumeNotFound)
     }
 
-    async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
+    async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
         let local_disks = self.local_disks_for_pools().await;
         if local_disks.is_empty() {
             return Err(Error::ErasureWriteQuorum);
@@ -562,7 +562,10 @@ impl PeerS3Client for LocalPeerS3Client {
         let mut futures = Vec::with_capacity(local_disks.len());
 
         for disk in local_disks.iter() {
-            futures.push(disk.delete_volume(bucket));
+            // Non-force delete refuses a non-empty bucket (VolumeNotEmpty), which
+            // the recreate loop below turns into BucketNotEmpty; only an explicit
+            // force delete removes recursively (backlog#799 B1).
+            futures.push(disk.delete_volume(bucket, opts.force));
         }
 
         let results = join_all(futures).await;
@@ -1036,9 +1039,19 @@ pub(crate) async fn heal_bucket_local_on_disks(
             futures.push(async move {
                 match disk {
                     Some(disk) => {
-                        info!("will call delete_volume, volume: {}", bucket);
-                        let _ = disk.delete_volume(&bucket).await;
-                        None
+                        // Non-force: a bucket that still holds object data refuses
+                        // deletion (VolumeNotEmpty) instead of being recursively
+                        // wiped, so a misclassified "dangling" bucket cannot lose
+                        // data (backlog#799 B1). Surface that refusal instead of
+                        // discarding it — it signals the bucket is not dangling.
+                        match disk.delete_volume(&bucket, false).await {
+                            Ok(()) => None,
+                            Err(Error::VolumeNotEmpty) => {
+                                warn!("heal declined to remove non-empty bucket {bucket} (not dangling)");
+                                None
+                            }
+                            Err(e) => Some(e),
+                        }
                     }
                     None => Some(Error::DiskNotFound),
                 }

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -1339,7 +1339,7 @@ impl DiskAPI for RemoteDisk {
     }
 
     #[tracing::instrument(skip(self))]
-    async fn delete_volume(&self, volume: &str) -> Result<()> {
+    async fn delete_volume(&self, volume: &str, force_delete: bool) -> Result<()> {
         debug!(
             event = EVENT_REMOTE_DISK_RPC,
             component = LOG_COMPONENT_ECSTORE,
@@ -1360,6 +1360,7 @@ impl DiskAPI for RemoteDisk {
                 let request = Request::new(DeleteVolumeRequest {
                     disk: self.endpoint.to_string(),
                     volume: volume.to_string(),
+                    force: force_delete,
                 });
 
                 let response = client.delete_volume(request).await?.into_inner();