diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs
index ab977e19a..658f2e4bd 100644
--- a/crates/ecstore/src/set_disk/core/io_primitives.rs
+++ b/crates/ecstore/src/set_disk/core/io_primitives.rs
@@ -54,7 +54,7 @@ use crate::set_disk::shard_source::ShardReadCost;
 use futures::stream::{FuturesUnordered, StreamExt};
 use metrics::counter;
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{HashMap, HashSet, VecDeque},
     future::Future,
     pin::Pin,
     sync::OnceLock,
@@ -3189,6 +3189,163 @@ impl SetDisks {
         Ok(true)
     }
 
+    /// Reclaim orphaned physical data directories under `bucket/object` that no
+    /// live version in the object's `xl.meta` references any longer.
+    ///
+    /// Before #3510, an unversioned overwrite leaked one UUID-named data dir per
+    /// PUT. The write path now cleans up going forward, but pre-existing strays
+    /// stay on disk forever: `heal`'s dangling logic only removes *whole* objects
+    /// whose data is missing, never surplus data dirs of an otherwise-healthy
+    /// object. This closes that gap so the scanner/heal sweep can recover the
+    /// leaked space automatically (issues #3231, #3191).
+    ///
+    /// Safety — fail closed:
+    /// * The set of referenced data dirs is the UNION of `get_data_dirs()` across
+    ///   every online disk's `xl.meta`, so a dir named by *any* replica is kept.
+    /// * If a disk holds the object directory but its `xl.meta` is missing or
+    ///   unparsable, the object is treated as degraded and NOTHING is removed —
+    ///   the unreadable copy could be the only one naming a live data dir, and a
+    ///   heal must run first.
+    /// * Only subdirectories whose names parse as a UUID are ever considered;
+    ///   removal is non-recursive-safe via a recursive delete of the full stray
+    ///   data-dir path only.
+    ///
+    /// Returns the number of stray data directories removed across the set. This
+    /// is best-effort maintenance: individual delete failures are logged and
+    /// skipped rather than propagated.
+    pub(crate) async fn reclaim_orphan_data_dirs(&self, bucket: &str, object: &str) -> disk::error::Result<usize> {
+        let disks = self.get_disks_internal().await;
+
+        // Phase 1 (read-only): build the referenced-data-dir union and record the
+        // physical UUID subdirectories present on each disk. Abort on any degraded
+        // copy so a healable object is never stripped of a referenced data dir.
+        let mut referenced: HashSet<Uuid> = HashSet::new();
+        let mut per_disk_dirs: Vec<(usize, Vec<Uuid>)> = Vec::new();
+        let mut healthy_metas = 0usize;
+
+        for (i, disk) in disks.iter().enumerate() {
+            let Some(disk) = disk else { continue };
+
+            let entries = match disk.list_dir("", bucket, object, 0).await {
+                Ok(entries) => entries,
+                // No object directory on this disk: nothing to reclaim here.
+                Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => continue,
+                Err(err) => return Err(err),
+            };
+
+            // Collect the UUID-named subdirectories: these are the physical data
+            // dirs. Non-directory entries (xl.meta) and non-UUID names are ignored.
+            let mut physical = Vec::new();
+            for entry in &entries {
+                let Some(name) = entry.strip_suffix(SLASH_SEPARATOR) else { continue };
+                if let Ok(uuid) = Uuid::parse_str(name)
+                    && !uuid.is_nil()
+                {
+                    physical.push(uuid);
+                }
+            }
+
+            // Read and parse this replica's metadata. A directory that carries data
+            // dirs but no readable xl.meta is degraded — fail closed.
+            let meta_path = path_join_buf(&[object, STORAGE_FORMAT_FILE]);
+            let buf = match disk.read_metadata(bucket, &meta_path).await {
+                Ok(buf) => buf,
+                Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {
+                    if physical.is_empty() {
+                        // Bare directory with no data dirs and no metadata: leave it
+                        // to the orphan-dir / dangling-object heal paths.
+                        continue;
+                    }
+                    warn!(
+                        target: "rustfs_ecstore::set_disk",
+                        bucket, object,
+                        "reclaim_orphan_data_dirs: aborting, data dirs present without xl.meta on a disk"
+                    );
+                    return Ok(0);
+                }
+                Err(err) => return Err(err),
+            };
+
+            let meta = match FileMeta::load(&buf) {
+                Ok(meta) => meta,
+                Err(err) => {
+                    warn!(
+                        target: "rustfs_ecstore::set_disk",
+                        bucket, object, error = %err,
+                        "reclaim_orphan_data_dirs: aborting, unparsable xl.meta on a disk"
+                    );
+                    return Ok(0);
+                }
+            };
+
+            match meta.get_data_dirs() {
+                Ok(dirs) => referenced.extend(dirs.into_iter().flatten().filter(|d| !d.is_nil())),
+                Err(err) => {
+                    warn!(
+                        target: "rustfs_ecstore::set_disk",
+                        bucket, object, error = %err,
+                        "reclaim_orphan_data_dirs: aborting, could not decode data dirs from xl.meta"
+                    );
+                    return Ok(0);
+                }
+            }
+
+            healthy_metas += 1;
+            if !physical.is_empty() {
+                per_disk_dirs.push((i, physical));
+            }
+        }
+
+        // No healthy metadata anywhere: this is not a live object, so surplus dirs
+        // (if any) belong to the dangling-object heal path, not here.
+        if healthy_metas == 0 {
+            return Ok(0);
+        }
+
+        // Phase 2: delete every physical data dir not referenced by the union.
+        let mut removed = 0usize;
+        for (i, physical) in per_disk_dirs {
+            let Some(disk) = disks[i].as_ref() else { continue };
+            for dir in physical {
+                if referenced.contains(&dir) {
+                    continue;
+                }
+                let stray = format!("{object}/{dir}");
+                match disk
+                    .delete(
+                        bucket,
+                        &stray,
+                        DeleteOptions {
+                            recursive: true,
+                            immediate: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                {
+                    Ok(()) => {
+                        removed += 1;
+                        info!(
+                            target: "rustfs_ecstore::set_disk",
+                            bucket, object, data_dir = %dir,
+                            "reclaim_orphan_data_dirs: removed orphaned data directory"
+                        );
+                    }
+                    Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
+                    Err(err) => {
+                        warn!(
+                            target: "rustfs_ecstore::set_disk",
+                            bucket, object, data_dir = %dir, error = %err,
+                            "reclaim_orphan_data_dirs: failed to remove orphaned data directory"
+                        );
+                    }
+                }
+            }
+        }
+
+        Ok(removed)
+    }
+
     pub(in crate::set_disk) async fn check_write_precondition(
         &self,
         bucket: &str,
diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs
index ae6823b50..69135fd9f 100644
--- a/crates/ecstore/src/set_disk/mod.rs
+++ b/crates/ecstore/src/set_disk/mod.rs
@@ -4487,6 +4487,151 @@ mod tests {
         );
     }
 
+    // Build an `xl.meta` under `object_dir` whose versions reference `data_dirs`
+    // (one Object version per data dir, each with its own version id).
+    async fn write_object_meta_with_data_dirs(object_dir: &std::path::Path, bucket: &str, object: &str, data_dirs: &[Uuid]) {
+        fs::create_dir_all(object_dir).await.expect("object dir should be created");
+        let mut meta = FileMeta::default();
+        for data_dir in data_dirs {
+            let mut fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 1);
+            fi.name = object.to_string();
+            fi.version_id = Some(Uuid::new_v4());
+            fi.data_dir = Some(*data_dir);
+            fi.size = 1;
+            fi.mod_time = Some(OffsetDateTime::now_utc());
+            meta.add_version(fi).expect("metadata should accept file info");
+        }
+        let buf = meta.marshal_msg().expect("metadata should encode");
+        fs::write(object_dir.join(STORAGE_FORMAT_FILE), buf)
+            .await
+            .expect("metadata should be written");
+    }
+
+    // #3231/#3191: a data dir on disk that no version references (a pre-#3510
+    // unversioned-overwrite leak) must be reclaimed, leaving the live one intact.
+    #[tokio::test]
+    async fn reclaim_orphan_data_dirs_removes_unreferenced_dir() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let live = Uuid::new_v4();
+        let orphan = Uuid::new_v4();
+
+        let object_dir = root.join("bucket").join("obj");
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
+        fs::create_dir_all(object_dir.join(live.to_string()))
+            .await
+            .expect("live data dir should be created");
+        fs::write(object_dir.join(live.to_string()).join("part.1"), b"live")
+            .await
+            .expect("live part should be written");
+        fs::create_dir_all(object_dir.join(orphan.to_string()))
+            .await
+            .expect("orphan data dir should be created");
+        fs::write(object_dir.join(orphan.to_string()).join("part.1"), b"stale")
+            .await
+            .expect("orphan part should be written");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let removed = set
+            .reclaim_orphan_data_dirs("bucket", "obj")
+            .await
+            .expect("reclaim should succeed");
+
+        assert_eq!(removed, 1, "exactly the unreferenced data dir should be removed");
+        assert!(object_dir.join(live.to_string()).exists(), "referenced data dir must be preserved");
+        assert!(!object_dir.join(orphan.to_string()).exists(), "orphaned data dir must be removed");
+        assert!(object_dir.join(STORAGE_FORMAT_FILE).exists(), "metadata must be preserved");
+    }
+
+    // Nothing to reclaim when every physical data dir is still referenced.
+    #[tokio::test]
+    async fn reclaim_orphan_data_dirs_keeps_referenced_dir() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let live = Uuid::new_v4();
+
+        let object_dir = root.join("bucket").join("obj");
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
+        fs::create_dir_all(object_dir.join(live.to_string()))
+            .await
+            .expect("live data dir should be created");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let removed = set
+            .reclaim_orphan_data_dirs("bucket", "obj")
+            .await
+            .expect("reclaim should succeed");
+
+        assert_eq!(removed, 0, "no data dir should be removed");
+        assert!(object_dir.join(live.to_string()).exists(), "referenced data dir must be preserved");
+    }
+
+    // Fail closed: a data dir present without a readable xl.meta is degraded, and
+    // must never be removed (a heal has to run first).
+    #[tokio::test]
+    async fn reclaim_orphan_data_dirs_aborts_when_meta_missing() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let stray = Uuid::new_v4();
+
+        let object_dir = root.join("bucket").join("obj");
+        fs::create_dir_all(object_dir.join(stray.to_string()))
+            .await
+            .expect("data dir should be created");
+        fs::write(object_dir.join(stray.to_string()).join("part.1"), b"data")
+            .await
+            .expect("part should be written");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let removed = set
+            .reclaim_orphan_data_dirs("bucket", "obj")
+            .await
+            .expect("reclaim should succeed");
+
+        assert_eq!(removed, 0, "must not remove data dirs when metadata is absent");
+        assert!(
+            object_dir.join(stray.to_string()).exists(),
+            "degraded object's data dir must be preserved"
+        );
+    }
+
+    // Cross-replica union: a data dir referenced by ANOTHER disk's xl.meta must be
+    // kept even where the local replica does not name it.
+    #[tokio::test]
+    async fn reclaim_orphan_data_dirs_keeps_dir_referenced_by_other_replica() {
+        let (dir0, disk0) = make_single_local_disk().await;
+        let (dir1, disk1) = make_single_local_disk().await;
+        let shared = Uuid::new_v4();
+
+        // disk0: meta references only a different dir, but physically holds `shared`.
+        let other = Uuid::new_v4();
+        let obj0 = dir0.path().join("bucket").join("obj");
+        write_object_meta_with_data_dirs(&obj0, "bucket", "obj", &[other]).await;
+        fs::create_dir_all(obj0.join(other.to_string()))
+            .await
+            .expect("dir should be created");
+        fs::create_dir_all(obj0.join(shared.to_string()))
+            .await
+            .expect("dir should be created");
+
+        // disk1: meta references `shared`.
+        let obj1 = dir1.path().join("bucket").join("obj");
+        write_object_meta_with_data_dirs(&obj1, "bucket", "obj", &[shared]).await;
+        fs::create_dir_all(obj1.join(shared.to_string()))
+            .await
+            .expect("dir should be created");
+
+        let set = make_set_disks_with(vec![Some(disk0), Some(disk1)]).await;
+        let removed = set
+            .reclaim_orphan_data_dirs("bucket", "obj")
+            .await
+            .expect("reclaim should succeed");
+
+        assert_eq!(removed, 0, "a dir referenced by any replica must be kept");
+        assert!(obj0.join(shared.to_string()).exists(), "cross-referenced data dir must survive");
+        assert!(obj0.join(other.to_string()).exists(), "locally referenced data dir must survive");
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn test_acquire_dist_delete_object_locks_batch_succeeds_with_two_healthy_lockers() {
diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs
index f3607eacc..97d23d2a9 100644
--- a/crates/ecstore/src/set_disk/ops/heal.rs
+++ b/crates/ecstore/src/set_disk/ops/heal.rs
@@ -624,6 +624,20 @@ impl SetDisks {
 
                         record_capacity_scope_if_needed(None, &out_dated_disks);
 
+                        // The object is healthy here; sweep any data dirs left behind
+                        // by pre-#3510 unversioned overwrites, which the dangling paths
+                        // above never touch (issues #3231, #3191). Best effort — a
+                        // failure must not fail the heal.
+                        match self.reclaim_orphan_data_dirs(bucket, object).await {
+                            Ok(removed) if removed > 0 => {
+                                info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
+                            }
+                            Ok(_) => {}
+                            Err(e) => {
+                                warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
+                            }
+                        }
+
                         Ok((result, None))
                     }
                     Err(err) => Ok((result, Some(err))),