From edae1ac31afa738ab064427975e338b434f9a860 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 1 Nov 2026 13:43:19 -0800 Subject: [PATCH] volume: refuse a batch delete while a replica is offline BatchDelete now applies the replica check the HTTP delete path already uses: when the master lists fewer replicas than the volume's copy count, the delete is refused with 503 instead of applied to the replicas that remain. The caller retries, so the replica that was offline still gets the delete once it returns. Co-Authored-By: Claude Opus 5.5 --- weed/server/volume_grpc_batch_delete.go | 25 ++++ .../volume_grpc_batch_delete_replica_test.go | 109 ++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 weed/server/volume_grpc_batch_delete_replica_test.go diff --git a/weed/server/volume_grpc_batch_delete.go b/weed/server/volume_grpc_batch_delete.go index 06a2eeb3ea..1a3dc90655 100644 --- a/weed/server/volume_grpc_batch_delete.go +++ b/weed/server/volume_grpc_batch_delete.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/topology" ) func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) { @@ -24,6 +25,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B } now := uint64(time.Now().Unix()) + replicaErrs := make(map[needle.VolumeId]error) for _, fid := range req.FileIds { vid, id_cookie, err := operation.ParseFileId(fid) @@ -56,6 +58,14 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B } } + if err := vs.checkDeleteReplicas(volumeId, replicaErrs); err != nil { + resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ + FileId: fid, + Status: http.StatusServiceUnavailable, + Error: err.Error(), + }) + continue + } ecVolume, isEcVolume := vs.store.FindEcVolume(volumeId) if !req.SkipCookieCheck { cookie := n.Cookie @@ -145,3 +155,18 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B return resp, nil } + +// checkDeleteReplicas refuses a delete the master cannot route to every +// replica, as the HTTP delete path does, so an offline replica does not keep +// the needle once it returns. +func (vs *VolumeServer) checkDeleteReplicas(volumeId needle.VolumeId, checked map[needle.VolumeId]error) error { + if err, found := checked[volumeId]; found { + return err + } + var err error + if v := vs.store.GetVolume(volumeId); v != nil && v.ReplicaPlacement.GetCopyCount() > 1 { + _, err = topology.GetRemoteReplications(vs.store, vs.grpcDialOption, volumeId, vs.GetMaster) + } + checked[volumeId] = err + return err +} diff --git a/weed/server/volume_grpc_batch_delete_replica_test.go b/weed/server/volume_grpc_batch_delete_replica_test.go new file mode 100644 index 0000000000..fcdf2f29db --- /dev/null +++ b/weed/server/volume_grpc_batch_delete_replica_test.go @@ -0,0 +1,109 @@ +package weed_server + +import ( + "context" + "net" + "net/http" + "strings" + "sync" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/peer" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +type replicaLookupMaster struct { + master_pb.UnimplementedSeaweedServer + mu sync.Mutex + locations []*master_pb.Location +} + +func (m *replicaLookupMaster) LookupVolume(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + m.mu.Lock() + defer m.mu.Unlock() + resp := &master_pb.LookupVolumeResponse{} + for _, vid := range req.VolumeOrFileIds { + resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{ + VolumeOrFileId: vid, + Locations: m.locations, + }) + } + return resp, nil +} + +func (m *replicaLookupMaster) setLocations(locations ...*master_pb.Location) { + m.mu.Lock() + defer m.mu.Unlock() + m.locations = locations +} + +// A replica the master no longer lists would miss the delete and keep the +// needle when it returns, so the delete waits until every replica is back. +func TestBatchDelete_WaitsForEveryReplica(t *testing.T) { + const vid = needle.VolumeId(9) + const cookie = uint32(0x12345678) + + store := newVolumeCopyTestStore(t, t.TempDir()) + if err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "001", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + data := []byte("hello") + if _, err := store.WriteVolumeNeedle(vid, &needle.Needle{Id: 1, Cookie: types.Cookie(cookie), Data: data, Checksum: needle.NewCRC(data)}, true, false); err != nil { + t.Fatalf("write needle: %v", err) + } + + self := &master_pb.Location{Url: "127.0.0.1:0"} + peerReplica := &master_pb.Location{Url: "127.0.0.2:8080"} + master := &replicaLookupMaster{} + master.setLocations(self) + masterGrpcPort := serveGrpc(t, func(s *grpc.Server) { + master_pb.RegisterSeaweedServer(s, master) + }) + + vs := &VolumeServer{ + store: store, + guard: security.NewGuard(nil, "", 0, "", 0), + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + vs.setCurrentMaster(pb.NewServerAddress("127.0.0.1", masterGrpcPort-10000, masterGrpcPort)) + + ctx := peer.NewContext(context.Background(), &peer.Peer{ + Addr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 12345}, + }) + req := &volume_server_pb.BatchDeleteRequest{ + FileIds: []string{needle.NewFileId(vid, 1, cookie).String()}, + SkipCookieCheck: true, + } + + resp, err := vs.BatchDelete(ctx, req) + if err != nil { + t.Fatalf("BatchDelete: %v", err) + } + if len(resp.Results) != 1 { + t.Fatalf("got %d results, want 1", len(resp.Results)) + } + if got := resp.Results[0]; got.Status != http.StatusServiceUnavailable || !strings.Contains(got.Error, "replication copy count") { + t.Fatalf("with a replica offline: status %d error %q, want %d and a replica count error", got.Status, got.Error, http.StatusServiceUnavailable) + } + if _, err := store.ReadVolumeNeedle(vid, &needle.Needle{Id: 1}, nil, nil); err != nil { + t.Fatalf("needle deleted while a replica was offline: %v", err) + } + + master.setLocations(self, peerReplica) + resp, err = vs.BatchDelete(ctx, req) + if err != nil { + t.Fatalf("BatchDelete: %v", err) + } + if got := resp.Results[0]; got.Status != http.StatusAccepted || got.Error != "" { + t.Fatalf("with every replica back: status %d error %q, want %d", got.Status, got.Error, http.StatusAccepted) + } +}