diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index b9e9b75c8..17331627b 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/weed/worker/tasks/erasure_coding/detection.go @@ -719,8 +719,20 @@ func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.Cluster var deleteErrors []string for _, replica := range replicas { serverAddress := replica.ServerID + var isReadOnly bool err := operation.WithVolumeServerClient(false, pb.ServerAddress(serverAddress), clusterInfo.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + // Re-probe before deleting: only a still-readonly source replica is + // safe to remove. One that came back writable may have accepted + // writes the EC shards do not contain, so deleting it loses data. + status, statusErr := client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: metric.VolumeID}) + if statusErr != nil { + return statusErr + } + isReadOnly = status.GetIsReadOnly() + if !isReadOnly { + return nil + } _, deleteErr := client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{ VolumeId: metric.VolumeID, OnlyEmpty: false, @@ -731,6 +743,10 @@ func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.Cluster deleteErrors = append(deleteErrors, fmt.Sprintf("server %s: %v", serverAddress, err)) continue } + if !isReadOnly { + glog.Warningf("EC Detection: source replica for volume %d on %s is writable; not deleting (may hold writes the EC shards lack)", metric.VolumeID, serverAddress) + continue + } deleted++ glog.V(1).Infof("EC Detection: deleted orphan regular replica for volume %d on %s", metric.VolumeID, serverAddress) } diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index fc12021ca..ad7b64279 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -342,7 +342,10 @@ func (t *ErasureCodingTask) markReplicasReadonly(ctx context.Context) error { addr := loc.ServerAddress() err := operation.WithVolumeServerClient(false, addr, t.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { - _, e := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: t.volumeID}) + // Persist the readonly mark so a source-server restart during or + // after encoding cannot silently reopen the volume to writes that + // the EC shards would not contain. rollbackReadonly clears it. + _, e := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: t.volumeID, Persist: true}) return e }) if err != nil { diff --git a/weed/worker/tasks/erasure_coding/ec_task_orphan_cleanup_test.go b/weed/worker/tasks/erasure_coding/ec_task_orphan_cleanup_test.go new file mode 100644 index 000000000..b2bbd46b9 --- /dev/null +++ b/weed/worker/tasks/erasure_coding/ec_task_orphan_cleanup_test.go @@ -0,0 +1,94 @@ +package erasure_coding + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/admin/topology" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/worker/types" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// cleanupOrphanSourceReplicas must not delete a source replica that has come +// back writable: it may hold writes the EC shards do not contain. A still +// readonly replica is the genuine orphan and is deleted. +func TestCleanupOrphanSkipsWritableSourceReplica(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(955) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + clusterInfo := &types.ClusterInfo{ + ActiveTopology: buildSingleNodeTopology(t, clusterHarness.VolumeServerAddress(), volumeID), + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + metric := &types.VolumeHealthMetrics{VolumeID: volumeID, Collection: ""} + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + // Writable source replica: must be left intact. + deleted, err := cleanupOrphanSourceReplicas(ctx, clusterInfo, metric, 0) + require.NoError(t, err) + require.Equal(t, 0, deleted, "a writable source replica must not be deleted") + require.True(t, volumeExists(t, grpcClient, volumeID), "writable source volume must still exist") + + // Mark readonly: now it is a genuine orphan and must be deleted. + _, err = grpcClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: volumeID}) + require.NoError(t, err) + + deleted, err = cleanupOrphanSourceReplicas(ctx, clusterInfo, metric, 0) + require.NoError(t, err) + require.Equal(t, 1, deleted, "a readonly orphan source replica must be deleted") + require.False(t, volumeExists(t, grpcClient, volumeID), "readonly orphan source volume must be deleted") +} + +func volumeExists(t *testing.T, client volume_server_pb.VolumeServerClient, volumeID uint32) bool { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, err := client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID}) + return err == nil +} + +// buildSingleNodeTopology builds an ActiveTopology with one node (at serverAddr) +// holding the regular volume, so cleanupOrphanSourceReplicas resolves the +// replica to the live test server and its gRPC calls reach it. +func buildSingleNodeTopology(t *testing.T, serverAddr string, volumeID uint32) *topology.ActiveTopology { + t.Helper() + at := topology.NewActiveTopology(10) + topologyInfo := &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{ + Id: "rack1", + DataNodeInfos: []*master_pb.DataNodeInfo{{ + Id: serverAddr, + DiskInfos: map[string]*master_pb.DiskInfo{ + "hdd": { + DiskId: 0, + VolumeInfos: []*master_pb.VolumeInformationMessage{ + {Id: volumeID, Collection: "", DiskId: 0}, + }, + }, + }, + }}, + }}, + }}, + } + require.NoError(t, at.UpdateTopology(topologyInfo)) + return at +} diff --git a/weed/worker/tasks/erasure_coding/ec_task_test.go b/weed/worker/tasks/erasure_coding/ec_task_test.go index e439398c4..a8e809d46 100644 --- a/weed/worker/tasks/erasure_coding/ec_task_test.go +++ b/weed/worker/tasks/erasure_coding/ec_task_test.go @@ -2,9 +2,11 @@ package erasure_coding import ( "context" + "fmt" "io" "net/http" "os" + "path/filepath" "testing" "time" @@ -78,6 +80,47 @@ func TestCopyVolumeFilesToWorkerUsesCurrentCompactionRevision(t *testing.T) { require.Equal(t, int64(fileStatus.GetIdxFileSize()), idxInfo.Size()) } +// markReplicasReadonly must persist the readonly mark into the .vif so a +// source-server restart cannot silently reopen the volume to writes that the +// EC shards would not contain. +func TestMarkReplicasReadonlyPersists(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(954) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 3001, 0x4444DDDD) + resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("payload-for-readonly-persist")) + _ = framework.ReadAllAndClose(t, resp) + require.Equal(t, http.StatusCreated, resp.StatusCode) + + task := NewErasureCodingTask( + "ec-readonly-persist", + clusterHarness.VolumeServerAddress(), + volumeID, + "", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + require.NoError(t, task.markReplicasReadonly(ctx)) + + vifPath := filepath.Join(clusterHarness.BaseDir(), "volume", fmt.Sprintf("%d.vif", volumeID)) + vi, _, found, err := volume_info.MaybeLoadVolumeInfo(vifPath) + require.NoError(t, err) + require.True(t, found, "volume must have a .vif after a persisted readonly mark") + require.True(t, vi.ReadOnly, "markReplicasReadonly must persist ReadOnly=true into the .vif") +} + // The worker-local encode path must stamp the EC ratio and an encode identity // into the .vif, or the read guard is silently off for worker-encoded volumes. func TestGenerateEcShardsLocallyStampsEncodeIdentity(t *testing.T) {