volume: scan for on-disk EC shards when staging a decoded volume (#10465)

The staged-new-volume placement skipped a disk holding the vid's EC shards using only the in-memory ecVolumes map, missing a shard present on disk but not mounted. Also scan the candidate disk for <vid>.ecNN files, so the promise holds regardless of mount state.

Claude-Session: https://claude.ai/code/session_01Ks16jnt4S7gdDk8cheQ3xu

seaweed-volume/src/server/grpc_server.rs

@@ -1769,12 +1769,28 @@ impl VolumeServer for VolumeGrpcService {
                                 // so the scanner never half-loads a partial push.
                                 let want = DiskType::from_string(&info.disk_type);
                                 let staged_vid = VolumeId(info.volume_id);
-                                // Skip a disk already holding this vid's EC shards, so the
-                                // decoded .dat never lands in the same directory as a shard —
-                                // lets the caller target a shard host that has a spare disk.
+                                // Don't stage the decoded .dat onto a disk that holds a
+                                // shard of this vid. Check the mounted map and the on-disk
+                                // files, so an unmounted or orphan shard (on disk, absent
+                                // from the map) is caught too.
                                 match store.find_free_location_predicate(|l| {
-                                    l.disk_type == want
-                                        && !l.ec_volumes().any(|(v, _)| *v == staged_vid)
+                                    if l.disk_type != want {
+                                        return false;
+                                    }
+                                    if l.ec_volumes().any(|(v, _)| *v == staged_vid) {
+                                        return false;
+                                    }
+                                    let base = crate::storage::volume::volume_file_name(
+                                        &l.directory,
+                                        &info.collection,
+                                        staged_vid,
+                                    );
+                                    !(0..crate::storage::erasure_coding::ec_shard::MAX_SHARD_COUNT)
+                                        .any(|i| {
+                                            std::fs::metadata(format!("{}.ec{:02}", base, i))
+                                                .map(|m| m.len() > 0)
+                                                .unwrap_or(false)
+                                        })
                                 }) {
                                     Some(i) => {
                                         let dir = store.locations[i].directory.clone();

weed/server/volume_grpc_copy.go

@@ -593,6 +593,19 @@ func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream v
 	return nil
 }
 
+// diskHoldsEcShardFile reports whether dir contains any <vid>.ecNN shard file,
+// so a decoded <vid>.dat is never staged beside a shard. Unlike FindEcVolume,
+// it also catches shards present on disk but not mounted.
+func diskHoldsEcShardFile(dir, collection string, vid needle.VolumeId) bool {
+	base := erasure_coding.EcShardFileName(collection, dir, int(vid))
+	for i := 0; i < erasure_coding.MaxShardCount; i++ {
+		if fi, err := os.Stat(base + erasure_coding.ToExt(i)); err == nil && !fi.IsDir() && fi.Size() > 0 {
+			return true
+		}
+	}
+	return false
+}
+
 // ReceiveFile receives a file stream from client and writes it to storage
 func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_ReceiveFileServer) error {
 	if err := vs.CheckMaintenanceMode(); err != nil {
@@ -702,13 +715,17 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive
 					want := types.ToDiskType(fileInfo.DiskType)
 					stagedVid := needle.VolumeId(fileInfo.VolumeId)
 					loc := vs.store.FindFreeLocation(func(l *storage.DiskLocation) bool {
-						// Skip a disk that already holds this vid's EC shards, so the
-						// decoded .dat never lands in the same directory as a shard —
-						// lets the caller target a shard host that has a spare disk.
+						if l.DiskType != want {
+							return false
+						}
+						// Don't stage the decoded .dat onto a disk that holds a shard
+						// of this vid. Check the mounted map and the on-disk files, so
+						// an unmounted or orphan shard (on disk, absent from the map)
+						// is caught too.
 						if _, holds := l.FindEcVolume(stagedVid); holds {
 							return false
 						}
-						return l.DiskType == want
+						return !diskHoldsEcShardFile(l.Directory, fileInfo.Collection, stagedVid)
 					})
 					if loc == nil {
 						return stream.SendAndClose(&volume_server_pb.ReceiveFileResponse{