fix(ec): clear stale EC shards cluster-wide before re-encoding

ec.encode unmounts and deletes EC shards for the target volumes on every
node before regenerating: fatal for the shards the topology reports
(mounted leftovers), best-effort for the rest (a sweep that catches
unmounted failed-copy orphans). A down node is a no-op.

weed/shell/command_ec_common.go

@@ -536,6 +536,30 @@ func sourceServerDeleteEcShards(grpcDialOption grpc.DialOption, collection strin
 
 }
 
+// unmountAndDeleteEcShardsQuiet unmounts then deletes shards on one server in a
+// single connection, without the per-call logging the interactive helpers emit.
+// Used by the orphan sweep, which fans out to every node x volume and would
+// otherwise flood the shell with no-op lines.
+func unmountAndDeleteEcShardsQuiet(grpcDialOption grpc.DialOption, collection string, volumeId needle.VolumeId, location pb.ServerAddress, shardIds []erasure_coding.ShardId) error {
+	ids := erasure_coding.ShardIdsToUint32(shardIds)
+	return operation.WithVolumeServerClient(false, location, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		if _, err := volumeServerClient.VolumeEcShardsUnmount(context.Background(), &volume_server_pb.VolumeEcShardsUnmountRequest{
+			VolumeId: uint32(volumeId),
+			ShardIds: ids,
+		}); err != nil {
+			return fmt.Errorf("unmount: %w", err)
+		}
+		if _, err := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{
+			VolumeId:   uint32(volumeId),
+			Collection: collection,
+			ShardIds:   ids,
+		}); err != nil {
+			return fmt.Errorf("delete: %w", err)
+		}
+		return nil
+	})
+}
+
 func unmountEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceLocation pb.ServerAddress, toBeUnmountedShardIds []erasure_coding.ShardId) error {
 
 	fmt.Printf("unmount %d.%v from %s\n", volumeId, toBeUnmountedShardIds, sourceLocation)

weed/shell/command_ec_encode.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"regexp"
 	"sort"
+	"strconv"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -236,6 +237,12 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
 		return fmt.Errorf("failed to get volume locations for EC encoding: %w", err)
 	}
 
+	// Clear EC shards left by a previous failed/partial encode so a retry
+	// starts clean and never mixes two encode runs.
+	if err := clearPreexistingEcShards(commandEnv, topologyInfo, volumeIds, volumeIdToCollection, maxParallelization); err != nil {
+		return fmt.Errorf("clear pre-existing ec shards before encoding: %w", err)
+	}
+
 	// Build a map of (volumeId, serverAddress) -> freeVolumeCount.
 	// Key by dn.Address so it matches wdclient.Location.Url. In deployments
 	// where dn.Id is a short name (e.g. Kubernetes StatefulSet pod name)
@@ -340,6 +347,73 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m
 	return nil
 }
 
+// clearPreexistingEcShards removes EC shards and index files left over from a
+// previous (failed or partial) encode of the given volume ids, on every node
+// that still reports them, so a fresh encode regenerates from a clean slate.
+// Scans all disk types. The normal .dat/.idx — the source of truth for this
+// encode — is untouched; only orphaned EC artifacts are deleted.
+func clearPreexistingEcShards(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeIds []needle.VolumeId, volumeIdToCollection map[needle.VolumeId]string, maxParallelization int) error {
+	wanted := make(map[uint32]bool, len(volumeIds))
+	for _, vid := range volumeIds {
+		wanted[uint32(vid)] = true
+	}
+
+	// Note which (node, vid) pairs the topology already reports EC shards for:
+	// those are mounted leftovers and cleaning them is required (fatal on
+	// error). Every other (node, vid) is swept best-effort to catch UNMOUNTED
+	// orphans left by a failed copy — invisible to the heartbeat, so absent
+	// here. A node that is down or holds nothing is a harmless no-op; a node
+	// unreachable now also cannot receive this encode's new generation, so a
+	// surviving orphan there keeps its old identity and the read guard rejects
+	// it. Always delete the full shard-id range so a wider custom ratio's
+	// leftovers are covered too.
+	reportedKey := func(addr pb.ServerAddress, vid uint32) string {
+		return string(addr) + "\x00" + strconv.Itoa(int(vid))
+	}
+	reported := make(map[string]struct{})
+	var nodes []pb.ServerAddress
+	eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		addr := pb.NewServerAddressFromDataNode(dn)
+		nodes = append(nodes, addr)
+		for _, diskInfo := range dn.DiskInfos {
+			for _, ecInfo := range diskInfo.EcShardInfos {
+				if wanted[ecInfo.Id] {
+					reported[reportedKey(addr, ecInfo.Id)] = struct{}{}
+				}
+			}
+		}
+	})
+
+	allShardIds := make([]erasure_coding.ShardId, erasure_coding.MaxShardCount)
+	for i := range allShardIds {
+		allShardIds[i] = erasure_coding.ShardId(i)
+	}
+
+	if len(reported) > 0 {
+		fmt.Printf("clearing stale EC shards reported for %d (node,volume) pair(s) before regenerating...\n", len(reported))
+	}
+	ewg := NewErrorWaitGroup(maxParallelization)
+	for _, addr := range nodes {
+		for _, vid := range volumeIds {
+			fatal := false
+			if _, ok := reported[reportedKey(addr, uint32(vid))]; ok {
+				fatal = true
+			}
+			collection := volumeIdToCollection[vid]
+			ewg.Add(func() error {
+				if err := unmountAndDeleteEcShardsQuiet(commandEnv.option.GrpcDialOption, collection, vid, addr, allShardIds); err != nil {
+					if fatal {
+						return fmt.Errorf("clear stale ec shards for volume %d on %s: %w", vid, addr, err)
+					}
+					glog.V(1).Infof("orphan sweep: volume %d on %s: %v", vid, addr, err)
+				}
+				return nil
+			})
+		}
+	}
+	return ewg.Wait()
+}
+
 func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.VolumeId, diskType types.DiskType) error {
 	// Shard relocations from the preceding EC balance reach the master via
 	// volume-server heartbeats, so freshly distributed shards may not all be