From a3b68aa37cc8a161e232a7135d57dd389353cfaf Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 09:32:37 -0800 Subject: [PATCH] shell: confirm trim survivors still serve the volume before deleting (#12590) * shell: confirm trim survivors still serve the volume before deleting volume.fix.replication compared the doomed replica's index with every survivor and then deleted it unconditionally. A survivor that went away while the indexes were read still let the delete through, leaving fewer copies than the check had counted. Re-read each survivor's VolumeStatus after the index checks and skip the trim if any no longer has the volume. Co-Authored-By: Claude Opus 5.5 * shell: test a trim whose survivor goes away after its index read Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/shell/command_volume_fix_replication.go | 12 +++ .../command_volume_fix_replication_test.go | 76 +++++++++++++++++++ 2 files changed, 88 insertions(+) diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index 7cb38fd3ba..f1e2d69804 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -382,6 +382,18 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr break } } + // A survivor can go away while the indexes are read. + for _, replicaB := range replicas { + if checkErr != nil { + break + } + if replicaB.location.dataNode == replica.location.dataNode { + continue + } + if _, checkErr = readVolumeStatus(commandEnv.option.GrpcDialOption, pb.NewServerAddressFromDataNode(replicaB.location.dataNode), needle.VolumeId(replicaB.info.Id)); checkErr != nil { + fmt.Fprintf(writer, "skip deleting volume %d from %s: %s no longer serves it: %v\n", replica.info.Id, replica.location.dataNode.Id, replicaB.location.dataNode.Id, checkErr) + } + } if checkErr != nil { continue } diff --git a/weed/shell/command_volume_fix_replication_test.go b/weed/shell/command_volume_fix_replication_test.go index 5f9d711629..23e06938af 100644 --- a/weed/shell/command_volume_fix_replication_test.go +++ b/weed/shell/command_volume_fix_replication_test.go @@ -1,10 +1,18 @@ package shell import ( + "context" + "io" + "sync/atomic" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type testcase struct { @@ -731,3 +739,71 @@ func TestClassifyReplicaSet(t *testing.T) { }) } } + +type trimTestVolumeServer struct { + volume_server_pb.UnimplementedVolumeServerServer + afterIndex func() + gone atomic.Bool + deletes atomic.Int32 +} + +func (s *trimTestVolumeServer) CopyFile(_ *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) error { + if s.gone.Load() { + return status.Error(codes.NotFound, "volume not found") + } + if err := stream.Send(&volume_server_pb.CopyFileResponse{FileContent: needle_map.ToBytes(1, types.ToOffset(8), 4)}); err != nil { + return err + } + if s.afterIndex != nil { + s.afterIndex() + } + return nil +} + +func (s *trimTestVolumeServer) VolumeStatus(context.Context, *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) { + if s.gone.Load() { + return nil, status.Error(codes.NotFound, "volume not found") + } + return &volume_server_pb.VolumeStatusResponse{}, nil +} + +func (s *trimTestVolumeServer) VolumeDelete(context.Context, *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) { + s.deletes.Add(1) + return &volume_server_pb.VolumeDeleteResponse{}, nil +} + +func TestDeleteOneVolumeRechecksSurvivorsBeforeDelete(t *testing.T) { + for _, tc := range []struct { + name string + survivorGoes bool + wantDeleted int + }{ + {name: "survivor stays", wantDeleted: 1}, + {name: "survivor gone after its index read", survivorGoes: true, wantDeleted: 0}, + } { + t.Run(tc.name, func(t *testing.T) { + doomed, survivor := &trimTestVolumeServer{}, &trimTestVolumeServer{} + if tc.survivorGoes { + survivor.afterIndex = func() { survivor.gone.Store(true) } + } + commandEnv, _ := configureReplicationTestEnv() + replicas := []*VolumeReplica{ + {location: &location{dataNode: startConfigureReplicationTestVolumeServer(t, doomed, 1)}, info: &master_pb.VolumeInformationMessage{Id: 1}}, + {location: &location{dataNode: startConfigureReplicationTestVolumeServer(t, survivor, 1)}, info: &master_pb.VolumeInformationMessage{Id: 1}}, + } + pickDoomed := func([]*VolumeReplica, *super_block.ReplicaPlacement) *VolumeReplica { return replicas[0] } + + c := &commandVolumeFixReplication{} + deleted, err := c.deleteOneVolume(commandEnv, io.Discard, true, true, []uint32{1}, map[uint32][]*VolumeReplica{1: replicas}, pickDoomed) + if err != nil { + t.Fatalf("deleteOneVolume: %v", err) + } + if tc.survivorGoes && !survivor.gone.Load() { + t.Fatal("the survivor never served its index") + } + if deleted != tc.wantDeleted || int(doomed.deletes.Load()) != tc.wantDeleted { + t.Fatalf("deleted=%d VolumeDelete calls=%d, want %d", deleted, doomed.deletes.Load(), tc.wantDeleted) + } + }) + } +}