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 <chris.lu@gmail.com>

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

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 {

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)

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