From e919bec9d194b61aa07cef75f46b207dd687a019 Mon Sep 17 00:00:00 2001 From: ssshr-66 Date: Thu, 10 Sep 2026 14:33:05 +0800 Subject: [PATCH] fix(volume): Harden Volume Copy Validation and Failure Handling (#11252) * fix(volume): harden volume copy validation * fix(volume): use stream context for ReadVolumeFileStatus in VolumeCopy ReadVolumeFileStatus ran on context.Background() while the adjacent VolumeStatus call used stream.Context(), an inconsistency left over from the context revert in #11252. Use stream.Context() consistently so the source status check is cancelled with the VolumeCopy stream. * fix(volume): reserve destination before deleting existing replica FindFreeLocation now runs before DeleteVolume so a full target fails without destroying the existing replica. Previously, when the initial VolumeStatus check failed (advisory) but ReadVolumeFileStatus succeeded, the existing replica was deleted before a destination was reserved, risking data loss if no location had enough free space. Add a regression test verifying the existing replica survives when the destination is full and the initial status check fails. * fix(volume): count replaced replica slot in FindFreeLocation FindFreeLocation now accepts the volume being replaced so its slot is treated as available. Without this, a location at its MaxVolumeCount limit could not replace its sole replica even though deleting it would free the slot. VolumeCopy passes the volume ID so destination selection succeeds before the existing replica is deleted. Add TestVolumeCopyReplacesReplicaAtSlotLimit covering a single-slot location that must replace its only replica. --------- Co-authored-by: Chris Lu --- weed/server/volume_grpc_copy.go | 42 ++--- weed/server/volume_grpc_copy_verify_test.go | 190 ++++++++++++++++++++ weed/storage/store.go | 32 ++-- weed/storage/store_mount_validation_test.go | 4 +- 4 files changed, 234 insertions(+), 34 deletions(-) diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 33ecf72c5..a6755dd4b 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -60,7 +60,8 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre VolumeId: req.VolumeId, }) if err != nil { - return fmt.Errorf("read volume status failed, %w", err) + glog.Warningf("failed to read source volume %d status before copy; skip record count validation: %v", req.VolumeId, err) + sourceVolumeStatus = nil } volFileInfoResp, err = client.ReadVolumeFileStatus(stream.Context(), @@ -71,18 +72,6 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return fmt.Errorf("read volume file status failed, %w", err) } - // Source is reachable and holds the volume: only now is it safe to drop - // an existing local replica before overwriting its files. - if hasExistingVolume { - glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) - // keep remote data: the inbound copy carries a .vif that may point at - // the same cloud-tier object the existing volume references. - if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true); delErr != nil { - return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) - } - glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) - } - diskType := volFileInfoResp.DiskType if req.DiskType != "" { diskType = req.DiskType @@ -96,11 +85,23 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre location := vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool { return location.DiskType == types.ToDiskType(diskType) && location.AvailableSpace.Load() > neededSpace - }) + }, needle.VolumeId(req.VolumeId)) if location == nil { return fmt.Errorf("%s %s", util.ErrVolumeNoSpaceLeft, types.ToDiskType(diskType).ReadableString()) } + // Source is reachable and a destination is reserved: only now is it + // safe to drop an existing local replica before overwriting its files. + if hasExistingVolume { + glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) + // keep remote data: the inbound copy carries a .vif that may point at + // the same cloud-tier object the existing volume references. + if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true); delErr != nil { + return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) + } + glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) + } + dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId)) indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) @@ -203,8 +204,8 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre VolumeId: req.VolumeId, }) if statusErr != nil { - glog.Warningf("failed to read source volume %d status after copy; skip record count validation: %v", req.VolumeId, statusErr) - sourceVolumeStatusAfterCopy = nil + err = fmt.Errorf("read source volume %d status after copy failed: %w", req.VolumeId, statusErr) + return err } return nil @@ -242,16 +243,14 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } shouldValidateCopyCounts := copyCountsStable(sourceVolumeStatus, sourceVolumeStatusAfterCopy) - if sourceVolumeStatusAfterCopy == nil { - glog.V(1).Infof("source volume %d status was unavailable after copy; skip record count validation", req.VolumeId) - } else if !shouldValidateCopyCounts { + if !shouldValidateCopyCounts { glog.V(1).Infof("source volume %d changed during copy; skip record count validation", req.VolumeId) } // Load and validate the volume before announcing it to the master. A failed // validation is unloaded by the store without ever making the replica // routable. - err = vs.store.MountVolumeWithValidator(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { + err = vs.store.MountVolume(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { if !shouldValidateCopyCounts { return nil } @@ -340,6 +339,9 @@ func checkCopyCounts(origin *volume_server_pb.VolumeStatusResponse, targetFileCo } func copyCountsStable(before, after *volume_server_pb.VolumeStatusResponse) bool { + // A writable source may receive writes or deletions while its files are + // copied. In that case the before/after counts do not describe one stable + // snapshot, so strict target-count validation would report a false error. return before != nil && after != nil && before.FileCount == after.FileCount && before.FileDeletedCount == after.FileDeletedCount diff --git a/weed/server/volume_grpc_copy_verify_test.go b/weed/server/volume_grpc_copy_verify_test.go index ef10832ab..2c3150f76 100644 --- a/weed/server/volume_grpc_copy_verify_test.go +++ b/weed/server/volume_grpc_copy_verify_test.go @@ -2,6 +2,11 @@ package weed_server import ( "context" + "errors" + "fmt" + "os" + "strings" + "sync/atomic" "testing" "google.golang.org/grpc" @@ -16,6 +21,191 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) +type volumeCopyStatusServer struct { + volume_server_pb.UnimplementedVolumeServerServer + delegate *VolumeServer + failStatusCall int32 + statusErr error + statusCalls atomic.Int32 +} + +func (s *volumeCopyStatusServer) VolumeStatus(ctx context.Context, req *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) { + if s.statusCalls.Add(1) == s.failStatusCall { + return nil, s.statusErr + } + return s.delegate.VolumeStatus(ctx, req) +} + +func (s *volumeCopyStatusServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) { + return s.delegate.ReadVolumeFileStatus(ctx, req) +} + +func (s *volumeCopyStatusServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) error { + return s.delegate.CopyFile(req, stream) +} + +func newVolumeCopyTestStore(t *testing.T, dir string) *storage.Store { + t.Helper() + store := storage.NewStore( + grpc.WithTransportCredentials(insecure.NewCredentials()), + "127.0.0.1", 0, 0, "", "test-store", + []string{dir}, []int32{10}, []util.MinFreeSpace{{}}, + dir, storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, [][]string{nil}, + 0, stats.DefaultDiskIOProbeConfig(), + ) + store.Locations[0].AvailableSpace.Store(^uint64(0)) + t.Cleanup(store.Close) + return store +} + +func runVolumeCopyWithStatusFailure(t *testing.T, failStatusCall int32) (error, *storage.Store) { + t.Helper() + const vid = needle.VolumeId(43) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: failStatusCall, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + return err, targetStore +} + +func TestVolumeCopyContinuesWhenInitialStatusUnavailable(t *testing.T) { + err, targetStore := runVolumeCopyWithStatusFailure(t, 1) + if err != nil { + t.Fatalf("VolumeCopy should continue when the initial status is unavailable: %v", err) + } + if targetStore.GetVolume(43) == nil { + t.Fatal("copied volume was not mounted") + } +} + +func TestVolumeCopyFailsWhenFinalStatusUnavailable(t *testing.T) { + err, targetStore := runVolumeCopyWithStatusFailure(t, 2) + if err == nil || !strings.Contains(err.Error(), "status after copy") { + t.Fatalf("VolumeCopy error = %v, want final status error", err) + } + if targetStore.GetVolume(43) != nil { + t.Fatal("volume was mounted after final status validation failed") + } + select { + case message := <-targetStore.NewVolumesChan: + t.Fatalf("volume was announced after final status validation failed: %+v", message) + default: + } + dataBaseFileName := storage.VolumeFileName(targetStore.Locations[0].Directory, "", 43) + indexBaseFileName := storage.VolumeFileName(targetStore.Locations[0].IdxDirectory, "", 43) + for _, fileName := range []string{ + dataBaseFileName + ".dat", + indexBaseFileName + ".idx", + dataBaseFileName + ".vif", + dataBaseFileName + ".note", + } { + if _, statErr := os.Stat(fileName); !os.IsNotExist(statErr) { + t.Fatalf("copy artifact %s remains after final status validation failed: %v", fileName, statErr) + } + } +} + +func TestVolumeCopyKeepsExistingReplicaWhenDestinationFull(t *testing.T) { + const vid = needle.VolumeId(44) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: 1, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := targetStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add target volume: %v", err) + } + targetStore.Locations[0].AvailableSpace.Store(0) + + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + if err == nil { + t.Fatal("VolumeCopy should fail when no destination location is available") + } + if targetStore.GetVolume(vid) == nil { + t.Fatal("existing replica was destroyed before a destination was reserved") + } +} + +func TestVolumeCopyReplacesReplicaAtSlotLimit(t *testing.T) { + const vid = needle.VolumeId(45) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: 1, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := targetStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add target volume: %v", err) + } + targetStore.Locations[0].MaxVolumeCount = 1 + + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + if err != nil { + t.Fatalf("VolumeCopy should replace the replica at the slot limit: %v", err) + } + if targetStore.GetVolume(vid) == nil { + t.Fatal("replaced volume was not mounted") + } +} + // fakeVolumeCopyStream is a no-op VolumeServer_VolumeCopyServer; VolumeCopy // errors out before sending anything in this test. type fakeVolumeCopyStream struct { diff --git a/weed/storage/store.go b/weed/storage/store.go index 7d891fedc..a8983bb36 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -285,8 +285,12 @@ func (s *Store) findVolume(vid needle.VolumeId) *Volume { } return nil } -func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool) (ret *DiskLocation) { +func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool, replaceVid ...needle.VolumeId) (ret *DiskLocation) { max := int32(0) + var replace needle.VolumeId + if len(replaceVid) > 0 { + replace = replaceVid[0] + } for _, location := range s.Locations { if filterFn != nil && !filterFn(location) { continue @@ -295,6 +299,11 @@ func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool) (re continue } currentFreeCount := location.MaxVolumeCount - int32(location.VolumesLen()) + if replace != 0 { + if _, found := location.FindVolume(replace); found { + currentFreeCount++ + } + } currentFreeCount *= erasure_coding.DataShardsCount currentFreeCount -= int32(location.EcShardCount()) currentFreeCount /= erasure_coding.DataShardsCount @@ -966,24 +975,23 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { return persistErr } -func (s *Store) MountVolume(i needle.VolumeId, collection *string) error { - return s.mountVolume(i, collection, nil) +// MountVolume loads a volume and announces it after all optional validators +// succeed. A validator failure unloads the volume before returning the error, +// so an invalid newly copied replica is never announced to the master. +func (s *Store) MountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { + return s.mountVolume(i, collection, validators...) } -// MountVolumeWithValidator loads a volume, validates it before announcing it -// to the master, and unloads it when validation fails. This keeps an invalid -// newly copied replica out of the master's routable volume set. -func (s *Store) MountVolumeWithValidator(i needle.VolumeId, collection *string, validator func(*Volume) error) error { - return s.mountVolume(i, collection, validator) -} - -func (s *Store) mountVolume(i needle.VolumeId, collection *string, validator func(*Volume) error) error { +func (s *Store) mountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { for diskId, location := range s.Locations { if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind, collection); found == true { glog.V(0).Infof("mount volume %d", i) v := s.findVolume(i) v.diskId = uint32(diskId) // Set disk ID when mounting - if validator != nil { + for _, validator := range validators { + if validator == nil { + continue + } if err := validator(v); err != nil { if unloadErr := location.UnloadVolume(i); unloadErr != nil { return fmt.Errorf("%w; failed to unload volume %d after validation error: %v", err, i, unloadErr) diff --git a/weed/storage/store_mount_validation_test.go b/weed/storage/store_mount_validation_test.go index cf6b16eb0..64a7d7e9a 100644 --- a/weed/storage/store_mount_validation_test.go +++ b/weed/storage/store_mount_validation_test.go @@ -13,7 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) -func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { +func TestMountVolumeValidatorAnnouncesOnlyAfterValidation(t *testing.T) { dir := t.TempDir() const vid = needle.VolumeId(17) @@ -29,7 +29,7 @@ func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { t.Cleanup(store.Close) validationErr := errors.New("copy counts differ") - err = store.MountVolumeWithValidator(vid, nil, func(*Volume) error { + err = store.MountVolume(vid, nil, func(*Volume) error { select { case <-store.NewVolumesChan: t.Fatal("volume was announced before validation completed")