fix(ecstore): settle lease-deferred data-dir deletes before bucket removal

A streaming GET holds snapshot leases on the object's data directories,
and DeleteObjects defers their physical cleanup until the leases are
released. A DeleteBucket issued inside that window passes the xl.meta
emptiness check but fails closed in the non-force delete_volume tree
removal on the leftover part files, returning BucketNotEmpty for a
logically empty bucket.

This is what intermittently failed the s3tests
test_encryption_sse_c_multipart_bad_download teardown in CI: the test
never reads its 30MiB GET body, so the server-side stream (and its
leases) stays alive until the connection drops, racing the teardown's
DeleteObjects + DeleteBucket sequence. With the body held open the
failure reproduces 5/5 locally; after this change it passes 20/20, and
the real s3-tests case passes 20 consecutive runs.

delete_volume now executes the registry-tracked pending deferred
deletions for the volume before removing the directory tree. Only data
dirs whose logical delete already committed are touched; unknown files
still fail closed with VolumeNotEmpty.

crates/ecstore/src/disk/local.rs

@@ -6641,6 +6641,49 @@ impl LocalDisk {
         let xl_path = object_dir.join(STORAGE_FORMAT_FILE);
         restore_delete_rollback_after_error(object_dir, &xl_path, Some(rollback_dir), volume, object, stage, err).await
     }
+
+    /// Execute every deferred data-dir deletion pending on `volume` right now,
+    /// even while snapshot leases are still held. Bucket deletion requires it:
+    /// a streaming reader defers the physical cleanup of an already-deleted
+    /// version, and a non-force `delete_volume` would otherwise fail closed
+    /// with `VolumeNotEmpty` on those remnants even though the bucket is
+    /// logically empty. The still-active readers keep their open descriptors;
+    /// 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;
+            registry
+                .entries
+                .iter_mut()
+                .filter(|(key, entry)| key.volume == volume && !entry.deleting && entry.pending_delete.is_some())
+                .map(|(key, entry)| {
+                    entry.deleting = true;
+                    (key.clone(), entry.pending_delete.clone().expect("filtered on Some"))
+                })
+                .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"
+                    );
+                }
+            }
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -9122,6 +9165,14 @@ impl DiskAPI for LocalDisk {
         let p = self.get_bucket_path(volume)?;
         let _volume_mutation_guard = os::disk_volume_mutation_lock(&self.root, volume).write_owned().await;
 
+        // A streaming reader's snapshot lease defers the physical cleanup of
+        // data dirs whose version delete already committed. Those remnants are
+        // logically deleted, so run the parked cleanups now instead of letting
+        // the non-force removal below fail closed on them (the s3-tests SSE-C
+        // teardown races exactly this way: DeleteObjects, then DeleteBucket
+        // while an abandoned GET body still pins the lease).
+        self.settle_pending_snapshot_deletes(volume).await;
+
         // Non-force removes empty directory remnants children-first with
         // non-recursive rmdir calls. A file that exists during the scan, or
         // appears before its parent is removed, fails closed with
@@ -15981,6 +16032,74 @@ mod test {
         assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound)));
     }
 
+    #[tokio::test]
+    async fn delete_volume_settles_lease_deferred_cleanup() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-lease-bucket-delete";
+        let object = "multipart_enc";
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let rollback_dir = Uuid::new_v4();
+        let data_path = path_join_buf(&[object, &data_dir.to_string()]);
+        let part = path_join_buf(&[&data_path, "part.1"]);
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &part, Bytes::from_static(b"payload"))
+            .await
+            .expect("shard should be written");
+        let fi = test_file_info(object, version_id, Some(data_dir), None);
+        disk.write_all(volume, &path_join_buf(&[object, STORAGE_FORMAT_FILE]), test_meta(fi.clone()).into())
+            .await
+            .expect("metadata should be written");
+
+        // An abandoned streaming GET pins the data dir with a snapshot lease.
+        let snapshot = disk
+            .acquire_snapshot_lease(volume, &data_path)
+            .await
+            .expect("snapshot lease should be acquired");
+        disk.delete_version(
+            volume,
+            object,
+            fi.clone(),
+            false,
+            DeleteOptions {
+                old_data_dir: Some(rollback_dir),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("version delete should commit metadata");
+        disk.delete(
+            volume,
+            &format!("{object}/{rollback_dir}"),
+            DeleteOptions {
+                recursive: true,
+                immediate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("version delete should schedule physical cleanup");
+
+        // The bucket is logically empty; a non-force volume delete must settle
+        // the deferred data-dir cleanup instead of failing with VolumeNotEmpty.
+        disk.delete_volume(volume, false)
+            .await
+            .expect("bucket delete must not observe lease-deferred remnants");
+        assert!(matches!(
+            disk.read_all(volume, &part).await,
+            Err(DiskError::FileNotFound | DiskError::VolumeNotFound)
+        ));
+
+        // The late lease release finds nothing pending and stays idempotent.
+        disk.release_snapshot_lease(volume, &data_path, snapshot)
+            .await
+            .expect("releasing the lease after bucket deletion should be a no-op");
+    }
+
     #[tokio::test]
     async fn version_delete_cleanup_intent_survives_local_disk_restart() {
         use tempfile::tempdir;