fix(ec): persist EC source readonly mark and skip writable replicas on orphan cleanup (#9950)

* fix(ec): persist the EC source replica readonly mark

markReplicasReadonly marked each regular replica readonly without
persisting it, so a source-server restart during or after encoding
silently reopened the volume to writes. Those writes are not in the EC
shards, and the later orphan-source cleanup would then delete the
replica, losing them. Send Persist:true so the mark survives a restart;
rollbackReadonly still clears it via VolumeMarkWritable on a failed
encode.

* fix(ec): don't delete a writable source replica during orphan cleanup

cleanupOrphanSourceReplicas issued VolumeDelete to every regular replica
once the EC shard set looked complete, without checking the replica's
current state. A replica that came back writable may hold writes the EC
shards do not contain, so deleting it loses data. Re-probe each replica
via VolumeStatus and skip any that is no longer readonly, logging a
warning instead of deleting.

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)
 	}

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 {

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
+}

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) {