fix(seaweed-volume): mirror prune of partial EC with sibling .dat (#9478)

Rust port of the same Store-level prune added to weed/storage. The
per-disk EC loader in disk_location.rs only checks for .dat in the same
disk as the EC shards, so an interrupted encode that leaves .ec?? + .ecx
on disk B while the source .dat sits on disk A is mounted as if it were
a distributed-EC layout. The volume server then heartbeats both a
regular replica and an EC shard for the same vid.

Sweep the store after per-disk loading and before the cross-disk
reconcile, dropping in-memory EcVolumes with fewer than DATA_SHARDS_COUNT
shards when a .dat for the same (collection, vid) exists on a sibling
disk, and remove all on-disk EC artefacts for them. The Rust heartbeat
path already diff-emits deletes from the next ec_volumes snapshot, so no
explicit delete-channel push is needed here.

Tests cover both the issue 9478 layout and a distributed-EC layout with
no .dat anywhere on the store, which must be left alone.

seaweed-volume/src/storage/disk_location.rs

@@ -304,7 +304,11 @@ impl DiskLocation {
     }
 
     /// Remove all EC-related files for a volume.
-    fn remove_ec_volume_files(&self, collection: &str, vid: VolumeId) {
+    /// `pub(crate)` so the Store-level cross-disk passes in
+    /// `store_ec_reconcile.rs` can call it to scrub partial EC artefacts
+    /// when a healthy `.dat` for the same vid lives on a sibling disk
+    /// (seaweedfs/seaweedfs#9478).
+    pub(crate) fn remove_ec_volume_files(&self, collection: &str, vid: VolumeId) {
         let base = volume_file_name(&self.directory, collection, vid);
         let idx_base = volume_file_name(&self.idx_directory, collection, vid);
         const MAX_SHARD_COUNT: usize = 32;

seaweed-volume/src/storage/store.rs

@@ -85,6 +85,17 @@ impl Store {
 
         self.locations.push(loc);
 
+        // First scrub partial EC artefacts left on one disk by an
+        // interrupted encode while the source .dat still lives on a
+        // sibling disk of the same store. The per-disk loader cannot
+        // see the sibling .dat and so loads the partial shards as if
+        // they were a distributed-EC layout, which makes the volume
+        // server heartbeat both a regular replica and an EC shard set
+        // for the same vid (seaweedfs/seaweedfs#9478). Running before
+        // the cross-disk reconcile keeps that pass from later
+        // re-loading shards we just cleaned up.
+        self.prune_incomplete_ec_with_sibling_dat();
+
         // After every disk has finished its per-disk EC scan, sweep
         // the store for shards that live on a disk without local index
         // files and load them by reaching across to a sibling disk's
@@ -106,6 +117,7 @@ impl Store {
                 tracing::error!("load_new_volumes error in {}: {}", loc.directory, e);
             }
         }
+        self.prune_incomplete_ec_with_sibling_dat();
         self.reconcile_ec_shards_across_disks();
     }
 

seaweed-volume/src/storage/store_ec_reconcile.rs

@@ -21,6 +21,7 @@ use std::fs;
 use tracing::{info, warn};
 
 use crate::storage::disk_location::{is_ec_shard_extension, parse_collection_volume_id_pub};
+use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
 use crate::storage::store::Store;
 use crate::storage::types::VolumeId;
 
@@ -136,6 +137,146 @@ impl Store {
         }
     }
 
+    /// Remove leftover EC artefacts on one disk when a healthy `.dat`
+    /// for the same `(collection, vid)` lives on a sibling disk of the
+    /// same store. Mirrors `Store.pruneIncompleteEcWithSiblingDat` in
+    /// `weed/storage/store_ec_reconcile.go` (seaweedfs/seaweedfs#9478).
+    ///
+    /// `DiskLocation::handle_found_ecx_file` only looks for `.dat` in
+    /// the same disk as the EC shards. In a multi-disk volume server an
+    /// interrupted EC encode can leave `.ec??` + `.ecx` on disk B while
+    /// the source `.dat` still lives on disk A. The per-disk loader
+    /// sees no local `.dat`, classifies the layout as distributed EC,
+    /// and mounts the partial shards. The volume server then heartbeats
+    /// both a regular replica and an EC shard for the same vid, and
+    /// master ends up with a contradictory view.
+    ///
+    /// Cleanup is gated on `shard_count < DATA_SHARDS_COUNT` so that a
+    /// deliberate "full local EC, `.dat` retained" layout split across
+    /// two disks (`.dat` on disk A, all 10+ shards on disk B) is left
+    /// alone — the per-disk loader already keeps that configuration on
+    /// a single disk, and pruning it here would be a behaviour
+    /// regression. Distributed EC volumes (no `.dat` on any disk of
+    /// this server) also fall through unchanged because the `.dat`
+    /// index never matches.
+    ///
+    /// We don't have to push anything to a deleted-shards channel here:
+    /// the Rust heartbeat path in `server/heartbeat.rs` diffs the
+    /// current `ec_volumes` snapshot against the previous one each
+    /// tick, so the first heartbeat after the prune naturally emits a
+    /// delete delta for whatever the per-disk pass had already
+    /// reported.
+    pub fn prune_incomplete_ec_with_sibling_dat(&mut self) {
+        if self.locations.len() < 2 {
+            return;
+        }
+
+        let dat_owners = self.index_dat_owners();
+        if dat_owners.is_empty() {
+            return;
+        }
+
+        // Snapshot under an immutable borrow so we can release it
+        // before taking the mutable borrow needed for cleanup.
+        struct Victim {
+            loc_idx: usize,
+            collection: String,
+            vid: VolumeId,
+            dat_dir: String,
+            shard_count: usize,
+        }
+        let mut victims: Vec<Victim> = Vec::new();
+        for (loc_idx, loc) in self.locations.iter().enumerate() {
+            for (vid, ev) in loc.ec_volumes() {
+                let shard_count = ev.shard_count();
+                if shard_count >= DATA_SHARDS_COUNT {
+                    continue;
+                }
+                let key = EcKey {
+                    collection: ev.collection.clone(),
+                    vid: *vid,
+                };
+                let Some(owner_idx) = dat_owners.get(&key) else {
+                    continue;
+                };
+                if *owner_idx == loc_idx {
+                    // Same-disk .dat is the job of
+                    // DiskLocation::validate_ec_volume during the
+                    // per-disk pass; don't second-guess it here.
+                    continue;
+                }
+                victims.push(Victim {
+                    loc_idx,
+                    collection: ev.collection.clone(),
+                    vid: *vid,
+                    dat_dir: self.locations[*owner_idx].directory.clone(),
+                    shard_count,
+                });
+            }
+        }
+
+        for v in victims {
+            warn!(
+                volume_id = v.vid.0,
+                collection = %v.collection,
+                directory = %self.locations[v.loc_idx].directory,
+                shard_count = v.shard_count,
+                required = DATA_SHARDS_COUNT,
+                dat_dir = %v.dat_dir,
+                "partial EC shards on this disk while a healthy .dat exists on a sibling disk; cleaning up leftover EC files (issue 9478)",
+            );
+            let loc = &mut self.locations[v.loc_idx];
+            if let Some(mut ec_vol) = loc.remove_ec_volume(v.vid) {
+                for _ in 0..ec_vol.shard_count() {
+                    crate::metrics::VOLUME_GAUGE
+                        .with_label_values(&[&ec_vol.collection, "ec_shards"])
+                        .dec();
+                }
+                // destroy() closes shard file handles before unlinking
+                // (matters on Windows) and removes .ecx / .ecj / .vif.
+                ec_vol.destroy();
+            }
+            // Also sweep any unmounted shard files (.ec00 .. .ec31)
+            // that the per-disk loader skipped — destroy() only walks
+            // the in-memory shards, but the disk may still hold others.
+            loc.remove_ec_volume_files(&v.collection, v.vid);
+        }
+    }
+
+    /// Returns, for every `(collection, vid)`, the index of the first
+    /// disk on this store that holds a `.dat` for it. Used by
+    /// `prune_incomplete_ec_with_sibling_dat` so it can decide whether
+    /// partial EC artefacts on another disk are leftovers of an
+    /// interrupted encode.
+    ///
+    /// Any `.dat` os::read_dir can see counts — including zero-byte
+    /// shells. Their mere presence means this volume was a regular
+    /// volume on this server at some point, which rules out the
+    /// "distributed EC, no .dat anywhere" reading that would justify
+    /// keeping the partial shards.
+    fn index_dat_owners(&self) -> HashMap<EcKey, usize> {
+        let mut owners: HashMap<EcKey, usize> = HashMap::new();
+        for (loc_idx, loc) in self.locations.iter().enumerate() {
+            let Ok(read) = fs::read_dir(&loc.directory) else {
+                continue;
+            };
+            for ent in read.flatten() {
+                if ent.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
+                    continue;
+                }
+                let name = ent.file_name().to_string_lossy().into_owned();
+                let Some(base) = name.strip_suffix(".dat") else {
+                    continue;
+                };
+                let Some((collection, vid)) = parse_collection_volume_id_pub(base) else {
+                    continue;
+                };
+                owners.entry(EcKey { collection, vid }).or_insert(loc_idx);
+            }
+        }
+        owners
+    }
+
     /// Build a `(collection, vid) -> EcxOwnerInfo` map of which disk
     /// owns the `.ecx` file. `.ecx` normally lives in `IdxDirectory`
     /// but may have been written into the data directory before
@@ -725,6 +866,144 @@ mod tests {
         assert!(ev.has_shard(0));
         assert!(ev.has_shard(1));
     }
+
+    /// Reproduces the issue 9478 layout for a single volume server:
+    /// disk A holds a healthy `.dat` for vid 122; disk B holds a single
+    /// `.ec01` plus `.ecx` / `.ecj` / `.vif` and no `.dat`. The per-disk
+    /// loader mounts the partial shard on disk B as if it were a
+    /// distributed-EC layout. The Store-level prune must unmount and
+    /// delete the leftover EC files on disk B while leaving disk A's
+    /// `.dat` untouched.
+    #[test]
+    fn test_prune_drops_partial_ec_when_sibling_disk_has_dat() {
+        let tmp = TempDir::new().unwrap();
+        let dat_dir = tmp.path().join("sdd");
+        let ec_dir = tmp.path().join("sdf");
+        std::fs::create_dir_all(&dat_dir).unwrap();
+        std::fs::create_dir_all(&ec_dir).unwrap();
+
+        let collection = "";
+        let vid = 122u32;
+
+        // Disk A (sdd): a non-trivial .dat plus the regular-volume
+        // sidecars. The .dat content doesn't matter for the prune —
+        // load_existing_volumes refuses to mount it because the
+        // superblock is absent — but its file name has to be present
+        // so index_dat_owners records this disk as the .dat owner.
+        let dat_path = dat_dir.join(format!("{}_{}.dat", collection, vid).trim_start_matches('_'));
+        std::fs::write(&dat_path, vec![0u8; 1024]).unwrap();
+
+        // Disk B (sdf): partial EC — one shard, plus .ecx / .ecj / .vif.
+        write_shard(ec_dir.to_str().unwrap(), collection, vid, 1);
+        write_index_files(ec_dir.to_str().unwrap(), collection, vid, 10, 4);
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dat_dir.to_str().unwrap(),
+                dat_dir.to_str().unwrap(),
+                100,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(0.0),
+                Vec::new(),
+            )
+            .unwrap();
+        store
+            .add_location(
+                ec_dir.to_str().unwrap(),
+                ec_dir.to_str().unwrap(),
+                100,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(0.0),
+                Vec::new(),
+            )
+            .unwrap();
+
+        // After add_location for sdf, the per-disk loader has mounted
+        // the partial shard AND the prune has run as part of
+        // add_location's epilogue. Confirm the prune removed it.
+        let ec_loc = &store.locations[1];
+        assert!(
+            ec_loc.find_ec_volume(VolumeId(vid)).is_none(),
+            "partial EC volume should have been unmounted by the prune",
+        );
+
+        let ec_base = ec_dir
+            .join(format!("{}_{}", collection, vid).trim_start_matches('_'))
+            .to_string_lossy()
+            .into_owned();
+        assert!(
+            !std::path::Path::new(&format!("{}.ec01", ec_base)).exists(),
+            "partial shard file should have been removed",
+        );
+        assert!(
+            !std::path::Path::new(&format!("{}.ecx", ec_base)).exists(),
+            ".ecx should have been removed",
+        );
+        assert!(
+            !std::path::Path::new(&format!("{}.ecj", ec_base)).exists(),
+            ".ecj should have been removed",
+        );
+
+        // The .dat on the sibling disk must survive — pruning the
+        // partial EC must never touch the healthy replica.
+        assert!(
+            dat_path.exists(),
+            "healthy .dat on sibling disk should be left alone",
+        );
+    }
+
+    /// A distributed EC volume — no `.dat` on any disk of this store —
+    /// must not be touched by the prune even when shard count is below
+    /// `DATA_SHARDS_COUNT`. Partial shard layouts are normal for
+    /// distributed EC.
+    #[test]
+    fn test_prune_leaves_distributed_ec_alone() {
+        let tmp = TempDir::new().unwrap();
+        let dir0 = tmp.path().join("data0");
+        let dir1 = tmp.path().join("data1");
+        std::fs::create_dir_all(&dir0).unwrap();
+        std::fs::create_dir_all(&dir1).unwrap();
+
+        let collection = "logs";
+        let vid = 7u32;
+
+        // Two shards on dir0 with .ecx, no .dat anywhere.
+        write_shard(dir0.to_str().unwrap(), collection, vid, 1);
+        write_shard(dir0.to_str().unwrap(), collection, vid, 2);
+        write_index_files(dir0.to_str().unwrap(), collection, vid, 10, 4);
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        for dir in [&dir0, &dir1] {
+            store
+                .add_location(
+                    dir.to_str().unwrap(),
+                    dir.to_str().unwrap(),
+                    100,
+                    DiskType::HardDrive,
+                    MinFreeSpace::Percent(0.0),
+                    Vec::new(),
+                )
+                .unwrap();
+        }
+
+        let ev = store.locations[0]
+            .find_ec_volume(VolumeId(vid))
+            .expect("distributed EC volume should still be mounted on dir0");
+        assert_eq!(
+            ev.shard_count(),
+            2,
+            "no .dat anywhere on this store means the partial shards must be kept as distributed EC",
+        );
+
+        let ec_base = dir0
+            .join(format!("{}_{}", collection, vid))
+            .to_string_lossy()
+            .into_owned();
+        assert!(std::path::Path::new(&format!("{}.ec01", ec_base)).exists());
+        assert!(std::path::Path::new(&format!("{}.ec02", ec_base)).exists());
+        assert!(std::path::Path::new(&format!("{}.ecx", ec_base)).exists());
+    }
 }
 
 /// Walk a disk's data directory and return the `.ec??` shard files