fix(ec): clear cross-server stale EC shards before re-distribute (#9478) (#9499)

* fix(ec): clear cross-server stale EC shards before re-distribute (#9478)

A previous failed encode leaves partial .ec?? shards mounted on
destination volume servers that are not the .dat owner. PR #9480 only
prunes when the .dat sits on a sibling disk of the SAME store, so the
cross-server case stays stuck: every retry trips
volume_grpc_copy.go:570's "ec volume %d is mounted; refusing overwrite"
guard and the scheduler loops.

Detection already lists existing EC shards as CleanupECShards sources;
plumb the shard ids through (ActiveTopology.GetECShardLocations,
TaskSourceSpec, TaskSource.shard_ids) and have the EC worker call
VolumeEcShardsUnmount + VolumeEcShardsDelete on each destination after
the local shard set is generated and before distributeEcShards. Skip
EC-shard sources in getReplicas so the post-encode VolumeDelete step
does not target destination-only nodes.

Integration test mounts a partial shard subset, asserts the
mounted-volume refusal, runs cleanupStaleEcShards, and asserts the
next ReceiveFile lands.

* chore(ec): tighten code comments in stale-shard cleanup

Drop issue-number refs from code comments and shorten the docstrings
on cleanupStaleEcShards / unmountAndDeleteEcShards / getReplicas plus
the new test file. Behavior unchanged.

* fix(ec): skip empty-ShardIds locations; dedupe getReplicas by node

GetECShardLocations dropped entries where ecShardMatchesCollection saw a
phantom info record with EcIndexBits=0 — without ShardIds, getReplicas
misread the resulting source as a regular replica and would have called
VolumeDelete on a destination-only node.

getReplicas now dedupes by Node since VolumeDelete is server-wide;
per-disk source rows on the same server collapse to one call.

* refactor(ec): use MaxShardCount and ShardBits in collectShardIdsForDisk

Drop the literal 32 bit-iteration bound for erasure_coding.MaxShardCount
and treat the EcIndexBits union as a ShardBits so Count() drives the
slice preallocation. Keeps the helper aligned with the rest of the EC
code and survives any future expansion of the shard-count ceiling.

weed/admin/topology/structs.go

@@ -113,10 +113,13 @@ type MultiDestinationPlan struct {
 	SuccessfulDCs  int                `json:"successful_dcs"`
 }
 
-// VolumeReplica represents a replica location with server and disk information
+// VolumeReplica represents a replica location with server and disk information.
+// ShardIds is populated only by GetECShardLocations — it lists the EC shards
+// the disk holds for the volume.
 type VolumeReplica struct {
-	ServerID   string `json:"server_id"`
-	DiskID     uint32 `json:"disk_id"`
-	DataCenter string `json:"data_center"`
-	Rack       string `json:"rack"`
+	ServerID   string   `json:"server_id"`
+	DiskID     uint32   `json:"disk_id"`
+	DataCenter string   `json:"data_center"`
+	Rack       string   `json:"rack"`
+	ShardIds   []uint32 `json:"shard_ids,omitempty"`
 }

weed/admin/topology/task_management.go

@@ -341,6 +341,7 @@ type TaskSourceSpec struct {
 	DataCenter    string             // Data center of the source server
 	Rack          string             // Rack of the source server
 	CleanupType   SourceCleanupType  // For EC: volume replica vs existing shards
+	ShardIds      []uint32           // For CleanupECShards: shard ids on the source disk to clear before re-distributing
 	StorageImpact *StorageSlotChange // Optional: manual override
 	EstimatedSize *int64             // Optional: manual override
 }

weed/admin/topology/topology_management.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 )
 
 // splitDiskInfoByPhysicalDisk returns one master_pb.DiskInfo per physical
@@ -333,7 +334,8 @@ func (at *ActiveTopology) GetVolumeLocations(volumeID uint32, collection string)
 	return replicas
 }
 
-// GetECShardLocations returns the disk locations for EC shards using O(1) lookup
+// GetECShardLocations returns the disk locations for EC shards using O(1) lookup.
+// Each VolumeReplica.ShardIds lists the shard ids on that disk.
 func (at *ActiveTopology) GetECShardLocations(volumeID uint32, collection string) []VolumeReplica {
 	at.mutex.RLock()
 	defer at.mutex.RUnlock()
@@ -345,22 +347,59 @@ func (at *ActiveTopology) GetECShardLocations(volumeID uint32, collection string
 
 	var ecShards []VolumeReplica
 	for _, diskKey := range diskKeys {
-		if disk, diskExists := at.disks[diskKey]; diskExists {
-			// Verify collection matches (since index doesn't include collection)
-			if at.ecShardMatchesCollection(disk, volumeID, collection) {
-				ecShards = append(ecShards, VolumeReplica{
-					ServerID:   disk.NodeID,
-					DiskID:     disk.DiskID,
-					DataCenter: disk.DataCenter,
-					Rack:       disk.Rack,
-				})
-			}
+		disk, diskExists := at.disks[diskKey]
+		if !diskExists {
+			continue
 		}
+		if !at.ecShardMatchesCollection(disk, volumeID, collection) {
+			continue
+		}
+		shardIds := collectShardIdsForDisk(disk, volumeID, collection)
+		if len(shardIds) == 0 {
+			// ecShardMatchesCollection saw an info entry but every
+			// EcIndexBits is zero — phantom shard record; emitting it
+			// would feed an EC-cleanup source with no shard ids and
+			// confuse the len(ShardIds) discriminator downstream.
+			continue
+		}
+		ecShards = append(ecShards, VolumeReplica{
+			ServerID:   disk.NodeID,
+			DiskID:     disk.DiskID,
+			DataCenter: disk.DataCenter,
+			Rack:       disk.Rack,
+			ShardIds:   shardIds,
+		})
 	}
 
 	return ecShards
 }
 
+// collectShardIdsForDisk unions every matching EcIndexBits on the disk and
+// expands the bitmap into shard ids, so multiple info entries for the same
+// volume don't produce duplicates.
+func collectShardIdsForDisk(disk *activeDisk, volumeID uint32, collection string) []uint32 {
+	if disk == nil || disk.DiskInfo == nil || disk.DiskInfo.DiskInfo == nil {
+		return nil
+	}
+	var bits erasure_coding.ShardBits
+	for _, ecShardInfo := range disk.DiskInfo.DiskInfo.EcShardInfos {
+		if ecShardInfo.Id != volumeID || ecShardInfo.Collection != collection {
+			continue
+		}
+		bits |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
+	}
+	if bits == 0 {
+		return nil
+	}
+	ids := make([]uint32, 0, bits.Count())
+	for id := uint32(0); id < erasure_coding.MaxShardCount; id++ {
+		if uint32(bits)&(1<<id) != 0 {
+			ids = append(ids, id)
+		}
+	}
+	return ids
+}
+
 // volumeMatchesCollection checks if a volume on a disk matches the given collection
 func (at *ActiveTopology) volumeMatchesCollection(disk *activeDisk, volumeID uint32, collection string) bool {
 	if disk.DiskInfo == nil || disk.DiskInfo.DiskInfo == nil {

weed/worker/tasks/erasure_coding/detection.go

@@ -273,6 +273,7 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 							DataCenter:  shard.DataCenter,
 							Rack:        shard.Rack,
 							CleanupType: topology.CleanupECShards,
+							ShardIds:    append([]uint32(nil), shard.ShardIds...),
 						})
 						duplicateCheck[key] = true
 					}
@@ -827,15 +828,15 @@ func convertTaskSourcesToProtobuf(sources []topology.TaskSourceSpec, volumeID ui
 			pbSource.EstimatedSize = uint64(*source.EstimatedSize)
 		}
 
-		// Set appropriate volume ID or shard IDs based on cleanup type
+		// Populated ShardIds is the wire-level marker that flags an
+		// EC-shard cleanup source; the worker routes it through
+		// cleanupStaleEcShards and skips it in getReplicas.
 		switch source.CleanupType {
 		case topology.CleanupVolumeReplica:
-			// This is a volume replica, use the actual volume ID
 			pbSource.VolumeId = volumeID
 		case topology.CleanupECShards:
-			// This is EC shards, also use the volume ID for consistency
 			pbSource.VolumeId = volumeID
-			// Note: ShardIds would need to be passed separately if we need specific shard info
+			pbSource.ShardIds = append([]uint32(nil), source.ShardIds...)
 		}
 
 		protobufSources = append(protobufSources, pbSource)

weed/worker/tasks/erasure_coding/ec_task.go

@@ -176,6 +176,16 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 		return fmt.Errorf("failed to generate EC shards: %v", err)
 	}
 
+	// Clear partial EC shards left over on destinations from a prior failed
+	// encode so distributeEcShards' ReceiveFile is not refused by the
+	// mounted-volume guard.
+	t.ReportProgressWithStage(55.0, "Clearing stale EC shards on destinations")
+	t.GetLogger().Info("Clearing stale EC shards on destinations")
+	if err := t.cleanupStaleEcShards(ctx); err != nil {
+		t.rollbackReadonly(ctx)
+		return fmt.Errorf("failed to clear stale EC shards on destinations: %v", err)
+	}
+
 	// Step 4: Distribute shards to destinations
 	t.ReportProgressWithStage(60.0, "Distributing EC shards to destinations")
 	t.GetLogger().Info("Distributing EC shards to destinations")
@@ -661,20 +671,116 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error {
 	return nil
 }
 
-// getReplicas extracts replica servers from unified sources
+// getReplicas extracts regular .dat replica servers from unified sources.
+// Sources with ShardIds set are EC-shard cleanup targets and must be skipped.
+// Per-disk source rows are deduped to one server entry — VolumeDelete is a
+// server-wide call.
 func (t *ErasureCodingTask) getReplicas() []string {
 	var replicas []string
+	seen := make(map[string]struct{})
 	for _, source := range t.sources {
-		// Only include volume replica sources (not EC shard sources)
-		// Assumption: VolumeId == 0 is considered invalid and should be excluded.
-		// If volume ID 0 is valid in some contexts, update this check accordingly.
-		if source.VolumeId > 0 {
-			replicas = append(replicas, source.Node)
+		if source.VolumeId == 0 || len(source.ShardIds) > 0 {
+			continue
 		}
+		if _, ok := seen[source.Node]; ok {
+			continue
+		}
+		seen[source.Node] = struct{}{}
+		replicas = append(replicas, source.Node)
 	}
 	return replicas
 }
 
+// cleanupStaleEcShards unmounts and deletes partial EC shards still mounted
+// on destinations from a previous failed encode. Safe by ordering: runs
+// after the source .dat is in the worker's workdir and a full local shard
+// set is generated. Per-destination errors are aggregated, not short-circuited.
+func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error {
+	if len(t.sources) == 0 {
+		return nil
+	}
+
+	// Union shard ids per destination node — volume-server cleanup walks
+	// every DiskLocation, so per-disk source rows collapse to one RPC.
+	perNode := make(map[string]map[uint32]struct{})
+	for _, source := range t.sources {
+		if source == nil || len(source.ShardIds) == 0 {
+			continue
+		}
+		shardSet, ok := perNode[source.Node]
+		if !ok {
+			shardSet = make(map[uint32]struct{})
+			perNode[source.Node] = shardSet
+		}
+		for _, shardID := range source.ShardIds {
+			shardSet[shardID] = struct{}{}
+		}
+	}
+	if len(perNode) == 0 {
+		return nil
+	}
+
+	var cleanupErrors []string
+	for node, shardSet := range perNode {
+		shardIds := make([]uint32, 0, len(shardSet))
+		for id := range shardSet {
+			shardIds = append(shardIds, id)
+		}
+
+		t.GetLogger().WithFields(map[string]interface{}{
+			"volume_id":   t.volumeID,
+			"destination": node,
+			"shard_ids":   shardIds,
+		}).Info("Clearing stale EC shards on destination before re-distribute")
+
+		if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, shardIds); err != nil {
+			cleanupErrors = append(cleanupErrors, fmt.Sprintf("%s: %v", node, err))
+			t.GetLogger().WithFields(map[string]interface{}{
+				"volume_id":   t.volumeID,
+				"destination": node,
+				"shard_ids":   shardIds,
+				"error":       err.Error(),
+			}).Error("Failed to clear stale EC shards on destination")
+		}
+	}
+
+	if len(cleanupErrors) > 0 {
+		return fmt.Errorf("stale EC shard cleanup failed on %d destination(s): %s",
+			len(cleanupErrors), strings.Join(cleanupErrors, "; "))
+	}
+	return nil
+}
+
+// unmountAndDeleteEcShards unmounts then deletes the named shards on one
+// destination. Unmount must precede delete (delete requires the shard be
+// unmounted); both RPCs are idempotent against missing shards.
+func unmountAndDeleteEcShards(
+	ctx context.Context,
+	dialOption grpc.DialOption,
+	destination string,
+	volumeID uint32,
+	collection string,
+	shardIds []uint32,
+) error {
+	return operation.WithVolumeServerClient(false, pb.ServerAddress(destination), dialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			if _, err := client.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{
+				VolumeId: volumeID,
+				ShardIds: shardIds,
+			}); err != nil {
+				return fmt.Errorf("unmount: %w", err)
+			}
+			if _, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
+				VolumeId:   volumeID,
+				Collection: collection,
+				ShardIds:   shardIds,
+			}); err != nil {
+				return fmt.Errorf("delete: %w", err)
+			}
+			return nil
+		})
+}
+
 // verifyDatIdxConsistency checks that all .idx entries reference data within the
 // .dat file. Since .dat and .idx are copied as separate network transfers, the
 // .idx may have entries from writes that landed after the .dat was copied.

weed/worker/tasks/erasure_coding/ec_task_stale_shard_cleanup_test.go

@@ -0,0 +1,225 @@
+package erasure_coding
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"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/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// Reproduces a stuck re-encode: partial EC shards mounted on a destination
+// from a previous failed encode cause ReceiveFile to refuse with the
+// mounted-volume guard. cleanupStaleEcShards must clear them so the next
+// ReceiveFile lands.
+func TestCleanupStaleEcShardsBeforeDistribute(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const (
+		volumeID   = uint32(9478)
+		collection = "ec-9478-xserver"
+	)
+
+	framework.AllocateVolume(t, grpcClient, volumeID, collection)
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 947800, 0x9478CAFE)
+	upResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("payload-for-cross-server-stale-ec-cleanup"))
+	_ = framework.ReadAllAndClose(t, upResp)
+	require.Equal(t, http.StatusCreated, upResp.StatusCode)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId: volumeID, Collection: collection,
+	})
+	require.NoError(t, err)
+
+	// Partial subset mimics a half-finished previous distribute: shards
+	// mounted on the destination with no .dat to anchor a same-store prune.
+	staleShards := []uint32{0, 1, 2}
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId: volumeID, Collection: collection,
+		ShardIds: staleShards,
+	})
+	require.NoError(t, err)
+
+	shardPath := makeTinyEcShardFile(t)
+
+	// Pre-cleanup: the mounted partial EC blocks ReceiveFile.
+	err = sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath)
+	require.Error(t, err, "expected ReceiveFile to be refused while EC volume is mounted")
+	require.True(t,
+		strings.Contains(err.Error(), "is mounted") ||
+			strings.Contains(err.Error(), "unmount before ReceiveFile"),
+		"expected refusal to name the mounted-volume guard, got: %v", err)
+
+	// ShardIds set marks this as an EC-shard cleanup source: cleanup will
+	// target it; getReplicas must skip it.
+	task := NewErasureCodingTask(
+		"stale-ec-xserver",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		collection,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.dataShards = erasure_coding.DataShardsCount
+	task.parityShards = erasure_coding.ParityShardsCount
+	task.sources = []*worker_pb.TaskSource{
+		{
+			Node:     clusterHarness.VolumeServerAddress(),
+			VolumeId: volumeID,
+			ShardIds: staleShards,
+		},
+	}
+
+	require.NoError(t, task.cleanupStaleEcShards(ctx))
+
+	_, infoErr := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: volumeID})
+	require.Error(t, infoErr, "EC volume should be gone after cleanupStaleEcShards")
+
+	require.NoError(t,
+		sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath),
+		"ReceiveFile must succeed after cleanup")
+
+	require.Empty(t, task.getReplicas(),
+		"EC-shard sources must not appear in replica delete list")
+}
+
+// Cleanup is a no-op when sources carry only the regular .dat replica.
+func TestCleanupStaleEcShardsSkipsRegularReplicas(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(9479)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	task := NewErasureCodingTask(
+		"no-stale-ec",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		"",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.sources = []*worker_pb.TaskSource{
+		{Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID},
+	}
+
+	require.NoError(t, task.cleanupStaleEcShards(ctx))
+
+	_, err := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.NoError(t, err, "regular volume must remain untouched")
+}
+
+// makeTinyEcShardFile writes a placeholder payload — the mounted-volume
+// guard fires before any content is consumed, so the bytes don't need to
+// be a real shard.
+func makeTinyEcShardFile(t *testing.T) string {
+	t.Helper()
+	p := filepath.Join(t.TempDir(), "shard.bin")
+	require.NoError(t, os.WriteFile(p, []byte("ec-shard-placeholder"), 0o600))
+	return p
+}
+
+// sendShardViaReceiveFile streams a shard file through the same ReceiveFile
+// gRPC the EC worker uses, returning the server's reply error verbatim.
+func sendShardViaReceiveFile(
+	ctx context.Context,
+	client volume_server_pb.VolumeServerClient,
+	volumeID uint32,
+	collection string,
+	shardID uint32,
+	filePath string,
+) error {
+	f, err := os.Open(filePath)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	info, err := f.Stat()
+	if err != nil {
+		return err
+	}
+
+	stream, err := client.ReceiveFile(ctx)
+	if err != nil {
+		return err
+	}
+
+	if err := stream.Send(&volume_server_pb.ReceiveFileRequest{
+		Data: &volume_server_pb.ReceiveFileRequest_Info{
+			Info: &volume_server_pb.ReceiveFileInfo{
+				VolumeId:   volumeID,
+				Ext:        erasure_coding.ToExt(int(shardID)),
+				Collection: collection,
+				IsEcVolume: true,
+				ShardId:    shardID,
+				FileSize:   uint64(info.Size()),
+			},
+		},
+	}); err != nil {
+		return err
+	}
+
+	buf := make([]byte, 32*1024)
+	for {
+		n, readErr := f.Read(buf)
+		if n > 0 {
+			if err := stream.Send(&volume_server_pb.ReceiveFileRequest{
+				Data: &volume_server_pb.ReceiveFileRequest_FileContent{
+					FileContent: buf[:n],
+				},
+			}); err != nil {
+				return err
+			}
+		}
+		if readErr == io.EOF {
+			break
+		}
+		if readErr != nil {
+			return readErr
+		}
+	}
+
+	resp, err := stream.CloseAndRecv()
+	if err != nil {
+		return err
+	}
+	if resp.Error != "" {
+		return &receiveFileServerError{msg: resp.Error}
+	}
+	return nil
+}
+
+type receiveFileServerError struct{ msg string }
+
+func (e *receiveFileServerError) Error() string { return e.msg }