From 2c1482f7a6fe0f12caa837014328671eb9141a24 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 14 May 2026 11:57:45 -0700 Subject: [PATCH] fix(ec): clear cross-server stale EC shards before re-distribute (#9478) (#9499) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 13 +- weed/admin/topology/task_management.go | 1 + weed/admin/topology/topology_management.go | 61 ++++- weed/worker/tasks/erasure_coding/detection.go | 9 +- weed/worker/tasks/erasure_coding/ec_task.go | 118 ++++++++- .../ec_task_stale_shard_cleanup_test.go | 225 ++++++++++++++++++ 6 files changed, 401 insertions(+), 26 deletions(-) create mode 100644 weed/worker/tasks/erasure_coding/ec_task_stale_shard_cleanup_test.go diff --git a/weed/admin/topology/structs.go b/weed/admin/topology/structs.go index 06903352e..ab4941888 100644 --- a/weed/admin/topology/structs.go +++ b/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"` } diff --git a/weed/admin/topology/task_management.go b/weed/admin/topology/task_management.go index 5fad4eda6..60b9dba86 100644 --- a/weed/admin/topology/task_management.go +++ b/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 } diff --git a/weed/admin/topology/topology_management.go b/weed/admin/topology/topology_management.go index 6b47695ab..ec9a7b74c 100644 --- a/weed/admin/topology/topology_management.go +++ b/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< 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. diff --git a/weed/worker/tasks/erasure_coding/ec_task_stale_shard_cleanup_test.go b/weed/worker/tasks/erasure_coding/ec_task_stale_shard_cleanup_test.go new file mode 100644 index 000000000..1e7f1c683 --- /dev/null +++ b/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 }