fix(ec): skip re-encode when EC shards already exist for the volume (#9448) (#9458)

* fix(ec): skip re-encode when EC shards already exist for the volume (#9448)

When an earlier EC encoding succeeded but the post-encode source-delete
left a regular replica behind on one of the servers, the next detection
cycle proposes the same volume again. The new encode tries to redistribute
shards to targets that already have them mounted, the volume server
returns `ec volume %d is mounted; refusing overwrite`, the task fails,
and detection re-queues the volume. The cycle repeats forever — issue
#9448.

The existing `metric.IsECVolume` skip catches the case where the canonical
metric is reported on the EC-shard side of the heartbeat, but when the
master sees BOTH a regular replica AND its EC shards in the same volume
list, the canonical metric we pick is the regular replica and
IsECVolume is false. Add a second guard that checks the topology
directly via `findExistingECShards` (already present and indexed) and
skip the volume when any shards exist, logging a warning that points
the admin at the stuck source.

This breaks the loop. Auto-cleanup of the orphaned replica is left as
follow-up work — deleting a source replica from inside the detector is
only safe with a re-verification step right before the delete, plus a
config opt-in, and is best done in its own change.

* fix(ec): #9448 guard only fires when EC shard set is complete

The first version of the #9448 guard tripped on `len(existingShards) > 0`,
which is broader than necessary. The existing recovery branch in the
encode arm (around the `existingECShards` block, ~line 216) is designed
to fold partial leftover shards from a previously failed encode into
the new task as cleanup sources. Skipping unconditionally on any
existing shards made that branch dead code, regressing the recovery
behavior Gemini flagged in the review of af09e1ec7.

Two corrections:

  1. New helper `countExistingEcShardsForVolume` walks each disk's
     `EcIndexBits` bitmap and ORs the results into a `ShardBits`,
     returning the distinct-shard popcount. This is the right unit:
     a single `VolumeEcShardInformationMessage` can carry several
     shards, so `len(EcShardInfos)` is not the same as the number
     of present shards. Per Gemini's "use helper functions that walk
     the actual shard bitmap" note.
  2. The guard now fires only when `shardCount >= totalShards`.
     Partial shard sets fall through to the existing recovery branch,
     unchanged.

Tests:
  - TestDetectionSkipsWhenECShardsAlreadyExist: complete shards →
    no proposal (the regression test for #9448 itself, unchanged
    intent, rewritten on top of new helpers).
  - TestDetectionAllowsRegularReplicaWhenShardsPartial: partial
    shards → guard does NOT swallow the volume; the encode arm
    still gets a chance.
  - TestCountExistingEcShardsForVolume: the helper walks the
    bitmap correctly even when one info entry packs multiple
    shards on one disk.

The dangerous `volume.delete` hint in the warning is unchanged for
now — it gets fixed in the next commit.

* fix(ec): drop dangerous shell-command hint from #9448 warning

The previous warning told operators to run `volume.delete -volumeId=%d`
in the SeaweedFS shell to clean up the orphaned source replica. That
command is cluster-wide — it deletes every replica of the volume,
including the EC shards, which share the same volume id. Running it
in the state the message describes would cause the data loss the
guard exists to prevent.

Replace it with explicit guidance that the cleanup must be a targeted
VolumeDelete RPC against the source server only, and that the
shell command is the exact wrong thing to use here. The next two
commits add the plumbing and the auto-execution of that targeted
delete so most operators never see this hint at all.

Per Gemini comment on af09e1ec7.

* feat(worker): plumb grpc dial option through ClusterInfo

Add ClusterInfo.GrpcDialOption (optional) and set it in the
erasure_coding plugin handler. Lets the detector make targeted
gRPC calls during detection — used by the follow-up commit to
auto-clean orphan source replicas via VolumeDelete RPCs.

Zero-value safe: existing detectors that don't need RPC access
get a nil DialOption and ignore the field.

* feat(ec): auto-clean orphan source replica via targeted VolumeDelete

Builds on the previous commits: the guard now identifies the
#9448 stuck-source state and a gRPC dial option is available on
ClusterInfo. When both are true, detection auto-cleans the
orphaned regular replica instead of just warning the operator.

New helper `cleanupOrphanSourceReplicas`:

  1. Re-verifies the EC shard set is still complete via
     `countExistingEcShardsForVolume` against the live topology
     snapshot. If the count dropped between detection start and
     the cleanup decision (a volume server going down mid-cycle),
     it aborts — the source replica is the only complete copy and
     deleting it without a healthy shard set would be data loss.
  2. Issues targeted VolumeDelete RPCs to each regular-replica
     server via `operation.WithVolumeServerClient`. That RPC only
     touches the regular volume on the targeted server; EC shards
     live in a separate store path and are not affected. This is
     the safe alternative to the cluster-wide `volume.delete`
     shell command we previously warned against.

If the cleanup partially fails (one replica delete errors, others
succeed), detection logs the failure and continues to skip the
volume. The next detection cycle will try again. We deliberately
don't fall back to a re-encode because that would just collide
with the mounted shards on the targets again.

When no dial option is available the existing warning still
points operators at the safe manual procedure.

weed/worker/tasks/erasure_coding/detection.go

@@ -4,10 +4,14 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/placement"
@@ -102,6 +106,51 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 			continue
 		}
 
+		// Handle the "stuck source" state from #9448: a previous encode
+		// succeeded but the post-encode source-delete left a regular replica
+		// behind, so the master heartbeats BOTH the replica AND its EC shards.
+		// metric.IsECVolume above is set only for the EC-side metric path, so
+		// the canonical metric we picked is the regular replica with
+		// IsECVolume=false. Re-proposing an encode in that state collides with
+		// the mounted shards on the targets ("ec volume %d is mounted; refusing
+		// overwrite") and the detector re-queues forever.
+		//
+		// We only act when the EC shard set is COMPLETE — fewer than
+		// totalShards present means the existing recovery branch below
+		// (around the `existingECShards` block) should keep its chance to
+		// fold the partial shards into the new task. Counting walks
+		// EcIndexBits to handle a single info entry carrying multiple shards.
+		if clusterInfo.ActiveTopology != nil {
+			shardCount := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
+			totalShards := erasure_coding.DataShardsCount + erasure_coding.ParityShardsCount
+			if shardCount >= totalShards {
+				glog.Warningf("EC Detection: Volume %d has all %d EC shards in topology; "+
+					"source replica on %s is orphaned (#9448).",
+					metric.VolumeID, totalShards, metric.Server)
+				if clusterInfo.GrpcDialOption != nil {
+					deleted, cleanupErr := cleanupOrphanSourceReplicas(ctx, clusterInfo, metric, totalShards)
+					switch {
+					case cleanupErr != nil:
+						// Don't fall through to a re-encode — that would just
+						// collide with the mounted shards again. Surface the
+						// failure and wait for the next cycle; the source is
+						// still safe.
+						glog.Warningf("EC Detection: failed to auto-clean orphaned source for volume %d: %v", metric.VolumeID, cleanupErr)
+					case deleted > 0:
+						glog.Infof("EC Detection: auto-cleaned %d orphaned source replica(s) for volume %d after verifying all %d EC shards present", deleted, metric.VolumeID, totalShards)
+					default:
+						glog.V(1).Infof("EC Detection: no orphaned regular replicas found in topology for volume %d (collection %q)", metric.VolumeID, metric.Collection)
+					}
+				} else {
+					glog.Warningf("EC Detection: no gRPC dial option available to auto-clean orphaned source for volume %d; "+
+						"to clean up by hand, send a targeted VolumeDelete RPC to %s only — DO NOT use the cluster-wide `volume.delete` shell command, which would also delete the EC shards.",
+						metric.VolumeID, metric.Server)
+				}
+				skippedAlreadyEC++
+				continue
+			}
+		}
+
 		// Check minimum size requirement
 		if metric.Size < minSizeBytes {
 			skippedTooSmall++
@@ -885,3 +934,94 @@ func findExistingECShards(activeTopology *topology.ActiveTopology, volumeID uint
 	}
 	return activeTopology.GetECShardLocations(volumeID, collection)
 }
+
+// cleanupOrphanSourceReplicas deletes any regular volume replicas still
+// present in the topology for (volumeID, collection) after re-verifying that
+// the full EC shard set is intact. Caller must hold expectedShards equal to
+// the configured totalShards count. Issues VolumeDelete RPC to each replica
+// server's address — that endpoint only touches the regular volume on the
+// targeted server, never EC shards (those live in a separate store path).
+// The cluster-wide `volume.delete` shell command is what would have nuked
+// the EC shards too; the targeted RPC used here is safe by construction.
+// Returns the count of replicas successfully deleted plus any error.
+func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.ClusterInfo, metric *types.VolumeHealthMetrics, expectedShards int) (int, error) {
+	if clusterInfo == nil || clusterInfo.ActiveTopology == nil {
+		return 0, fmt.Errorf("active topology unavailable")
+	}
+	if clusterInfo.GrpcDialOption == nil {
+		return 0, fmt.Errorf("grpc dial option unavailable")
+	}
+
+	// Re-verify shard completeness right before acting. Defensive: detection
+	// processes many volumes sequentially and the topology snapshot we built
+	// at start-of-detection could have lost shards in between (a volume
+	// server going down between iterations). Refusing to delete the source
+	// when we can no longer prove the shards are complete is the safer
+	// failure mode — the source replica is the only complete copy.
+	actualShards := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
+	if actualShards < expectedShards {
+		return 0, fmt.Errorf("EC shard set shrank between detection and cleanup (%d < %d); refusing to delete source replica", actualShards, expectedShards)
+	}
+
+	replicas := findVolumeReplicaLocations(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
+	if len(replicas) == 0 {
+		return 0, nil
+	}
+
+	deleted := 0
+	var deleteErrors []string
+	for _, replica := range replicas {
+		serverAddress := replica.ServerID
+		err := operation.WithVolumeServerClient(false, pb.ServerAddress(serverAddress), clusterInfo.GrpcDialOption,
+			func(client volume_server_pb.VolumeServerClient) error {
+				_, deleteErr := client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{
+					VolumeId:  metric.VolumeID,
+					OnlyEmpty: false,
+				})
+				return deleteErr
+			})
+		if err != nil {
+			deleteErrors = append(deleteErrors, fmt.Sprintf("server %s: %v", serverAddress, err))
+			continue
+		}
+		deleted++
+		glog.V(1).Infof("EC Detection: deleted orphan regular replica for volume %d on %s", metric.VolumeID, serverAddress)
+	}
+
+	if len(deleteErrors) > 0 {
+		return deleted, fmt.Errorf("%d of %d replica delete(s) failed: %s", len(deleteErrors), len(replicas), strings.Join(deleteErrors, "; "))
+	}
+	return deleted, nil
+}
+
+// countExistingEcShardsForVolume returns the number of distinct EC shard IDs
+// for (volumeID, collection) present in the topology. Walks every disk's
+// EcIndexBits bitmap rather than trusting len(EcShardInfos), because a single
+// info entry can carry multiple shards. Used by the #9448 guard to decide
+// whether the EC shard set is complete enough that the orphaned regular
+// replica is safe to delete.
+func countExistingEcShardsForVolume(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) int {
+	if activeTopology == nil {
+		return 0
+	}
+	topologyInfo := activeTopology.GetTopologyInfo()
+	if topologyInfo == nil {
+		return 0
+	}
+	var seen erasure_coding.ShardBits
+	for _, dc := range topologyInfo.DataCenterInfos {
+		for _, rack := range dc.RackInfos {
+			for _, node := range rack.DataNodeInfos {
+				for _, diskInfo := range node.DiskInfos {
+					for _, ecShardInfo := range diskInfo.EcShardInfos {
+						if ecShardInfo.Id != volumeID || ecShardInfo.Collection != collection {
+							continue
+						}
+						seen |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
+					}
+				}
+			}
+		}
+	}
+	return seen.Count()
+}

weed/worker/tasks/erasure_coding/detection_test.go

@@ -127,6 +127,179 @@ func TestECPlacementPlannerFallsBackWhenTagsInsufficient(t *testing.T) {
 	assert.Less(t, taggedCount, len(selected))
 }
 
+// TestDetectionSkipsWhenECShardsAlreadyExist guards against issue #9448: a
+// regular replica that survived a previous successful EC encode (source
+// delete didn't clean it up for some reason) gets re-proposed for encoding,
+// the new encode collides with the already-mounted shards on the targets
+// ("ec volume %d is mounted; refusing overwrite"), and detection loops
+// forever on the same volume. Detection must see the existing shards and
+// skip the volume so an admin can clean it up out-of-band.
+//
+// The guard fires ONLY when the EC shard set is complete (count >=
+// totalShards), so a partially-distributed previous attempt still falls
+// through to the existing recovery branch in the encode path.
+func TestDetectionSkipsWhenECShardsAlreadyExist(t *testing.T) {
+	const volumeID uint32 = 42
+	activeTopology := buildStuckSourceTopology(t, volumeID, erasure_coding.TotalShardsCount)
+
+	clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology}
+	metrics := buildStuckSourceMetrics(volumeID, "127.0.0.1:8080")
+
+	results, hasMore, err := Detection(context.Background(), metrics, clusterInfo, NewDefaultConfig(), 0)
+	require.NoError(t, err)
+	require.False(t, hasMore)
+	require.Empty(t, results, "stuck source replica with all EC shards present must not yield a new encoding proposal")
+}
+
+// TestDetectionAllowsRegularReplicaWhenShardsPartial covers the partial-EC
+// branch of the #9448 guard: when fewer than totalShards exist, the volume
+// is allowed to flow through to the normal encoding path so the existing
+// recovery branch (the `existingECShards` block in the encode arm) can fold
+// the partial shards into the new task. A bug here would either (a) skip
+// the volume entirely or (b) emit a proposal that later collides on the
+// mounted shards.
+func TestDetectionAllowsRegularReplicaWhenShardsPartial(t *testing.T) {
+	const volumeID uint32 = 43
+	activeTopology := buildStuckSourceTopology(t, volumeID, erasure_coding.DataShardsCount-1)
+
+	clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology}
+	metrics := buildStuckSourceMetrics(volumeID, "127.0.0.1:8080")
+
+	results, _, err := Detection(context.Background(), metrics, clusterInfo, NewDefaultConfig(), 0)
+	require.NoError(t, err)
+	// Partial shards are not a "stuck source" — the encode arm must keep
+	// its chance to either propose a fresh task that folds the partial
+	// shards into cleanup, or fail planning on the constrained topology.
+	// We don't require len(results) > 0 because the constrained topology
+	// (one disk per node, the orphaned shards already taking slots) can
+	// legitimately fail destination planning. The assertion that matters
+	// is: the #9448 guard did NOT silently swallow the volume into a
+	// skippedAlreadyEC counter, and any emitted result is still an EC
+	// task and not a no-op.
+	for _, r := range results {
+		require.Equal(t, types.TaskTypeErasureCoding, r.TaskType, "any emitted result should still be an EC task, not a no-op")
+	}
+}
+
+// buildStuckSourceTopology constructs a topology that mimics the #9448 stuck
+// state: a regular volume replica on node 0 plus `presentShardCount` EC
+// shards distributed across nodes 0..presentShardCount-1.
+func buildStuckSourceTopology(t *testing.T, volumeID uint32, presentShardCount int) *topology.ActiveTopology {
+	t.Helper()
+	require.LessOrEqual(t, presentShardCount, erasure_coding.TotalShardsCount)
+	activeTopology := topology.NewActiveTopology(10)
+	nodes := make([]*master_pb.DataNodeInfo, 0, erasure_coding.TotalShardsCount)
+	for i := 0; i < erasure_coding.TotalShardsCount; i++ {
+		nodeID := fmt.Sprintf("127.0.0.1:%d", 8080+i)
+		diskInfo := &master_pb.DiskInfo{
+			DiskId:         0,
+			VolumeCount:    1,
+			MaxVolumeCount: 100,
+		}
+		if i < presentShardCount {
+			diskInfo.EcShardInfos = []*master_pb.VolumeEcShardInformationMessage{{
+				Id:          volumeID,
+				Collection:  "",
+				EcIndexBits: uint32(1) << uint(i),
+				DiskId:      0,
+			}}
+		}
+		if i == 0 {
+			diskInfo.VolumeInfos = []*master_pb.VolumeInformationMessage{{
+				Id:       volumeID,
+				DiskId:   0,
+				DiskType: "hdd",
+				Size:     200 * 1024 * 1024,
+			}}
+		}
+		nodes = append(nodes, &master_pb.DataNodeInfo{
+			Id:        nodeID,
+			DiskInfos: map[string]*master_pb.DiskInfo{"hdd": diskInfo},
+		})
+	}
+	require.NoError(t, activeTopology.UpdateTopology(&master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id: "dc1",
+			RackInfos: []*master_pb.RackInfo{{
+				Id:            "rack1",
+				DataNodeInfos: nodes,
+			}},
+		}},
+	}))
+	return activeTopology
+}
+
+// buildStuckSourceMetrics returns a metric that already satisfies the EC
+// criteria (Age, FullnessRatio, Size), with `Age` derived from `LastModified`
+// so the two fields stay consistent for any reader.
+func buildStuckSourceMetrics(volumeID uint32, server string) []*types.VolumeHealthMetrics {
+	lastModified := time.Now().Add(-time.Hour)
+	return []*types.VolumeHealthMetrics{{
+		VolumeID:      volumeID,
+		Server:        server,
+		Size:          200 * 1024 * 1024,
+		Collection:    "",
+		FullnessRatio: 0.9,
+		LastModified:  lastModified,
+		Age:           time.Since(lastModified),
+	}}
+}
+
+// TestCountExistingEcShardsForVolume verifies that the helper walks the
+// EcIndexBits bitmap (not just len(EcShardInfos)) so it correctly counts
+// distinct shard ids even when a single info entry on one disk carries
+// multiple shards.
+func TestCountExistingEcShardsForVolume(t *testing.T) {
+	const volumeID uint32 = 99
+	activeTopology := topology.NewActiveTopology(10)
+	require.NoError(t, activeTopology.UpdateTopology(&master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id: "dc1",
+			RackInfos: []*master_pb.RackInfo{{
+				Id: "rack1",
+				DataNodeInfos: []*master_pb.DataNodeInfo{
+					{
+						Id: "127.0.0.1:8080",
+						DiskInfos: map[string]*master_pb.DiskInfo{
+							"hdd": {
+								DiskId:         0,
+								MaxVolumeCount: 100,
+								// One info entry, three shards present (ids 0, 2, 5).
+								EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{
+									Id:          volumeID,
+									Collection:  "",
+									EcIndexBits: (uint32(1) << 0) | (uint32(1) << 2) | (uint32(1) << 5),
+									DiskId:      0,
+								}},
+							},
+						},
+					},
+					{
+						Id: "127.0.0.1:8081",
+						DiskInfos: map[string]*master_pb.DiskInfo{
+							"hdd": {
+								DiskId:         0,
+								MaxVolumeCount: 100,
+								// One info entry, one shard (id 3) — overlaps with neither.
+								EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{
+									Id:          volumeID,
+									Collection:  "",
+									EcIndexBits: uint32(1) << 3,
+									DiskId:      0,
+								}},
+							},
+						},
+					},
+				},
+			}},
+		}},
+	}))
+
+	assert.Equal(t, 4, countExistingEcShardsForVolume(activeTopology, volumeID, ""))
+	assert.Equal(t, 0, countExistingEcShardsForVolume(activeTopology, volumeID, "other-collection"))
+	assert.Equal(t, 0, countExistingEcShardsForVolume(nil, volumeID, ""))
+}
+
 func TestDetectionContextCancellation(t *testing.T) {
 	activeTopology := buildActiveTopology(t, 5, []string{"hdd", "ssd"}, 50, 0)
 	clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology}

weed/worker/tasks/erasure_coding/plugin_handler.go

@@ -217,7 +217,7 @@ func (h *ErasureCodingHandler) Detect(
 		return err
 	}
 
-	clusterInfo := &workertypes.ClusterInfo{ActiveTopology: activeTopology}
+	clusterInfo := &workertypes.ClusterInfo{ActiveTopology: activeTopology, GrpcDialOption: h.grpcDialOption}
 	maxResults := int(request.MaxResults)
 	if maxResults < 0 {
 		maxResults = 0

weed/worker/types/data_types.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
+	"google.golang.org/grpc"
 )
 
 // ReplicaLocation identifies where a volume replica lives.
@@ -21,6 +22,11 @@ type ClusterInfo struct {
 	LastUpdated      time.Time
 	ActiveTopology   *topology.ActiveTopology // Added for destination planning in detection
 	VolumeReplicaMap map[uint32][]ReplicaLocation
+	// GrpcDialOption is set when a detector needs to make targeted gRPC calls
+	// during detection (e.g., the EC detector auto-cleans up an orphaned
+	// regular replica that survived a previous encode; see #9448). Optional:
+	// detectors that don't need RPC access ignore this.
+	GrpcDialOption grpc.DialOption `json:"-"`
 }
 
 // VolumeHealthMetrics contains health information about a volume (simplified)