[Volume] Validate record counts after volume copy (#11238)

* validate Volume Copy record counts

* Delete s3api_object_versioning_bench_test.go

* reply ai comments

weed/server/volume_grpc_copy.go

@@ -50,11 +50,20 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 	//   send .dat file
 	//   confirm size and timestamp
 	var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse
+	var sourceVolumeStatus *volume_server_pb.VolumeStatusResponse
+	var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse
 	var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string
 	var hasRemoteDatFile bool
 	err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 		var err error
-		volFileInfoResp, err = client.ReadVolumeFileStatus(context.Background(),
+		sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{
+			VolumeId: req.VolumeId,
+		})
+		if err != nil {
+			return fmt.Errorf("read volume status failed, %w", err)
+		}
+
+		volFileInfoResp, err = client.ReadVolumeFileStatus(stream.Context(),
 			&volume_server_pb.ReadVolumeFileStatusRequest{
 				VolumeId: req.VolumeId,
 			})
@@ -189,6 +198,15 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 			return fmt.Errorf("remove .note for volume %d: %w", req.VolumeId, noteErr)
 		}
 
+		var statusErr error
+		sourceVolumeStatusAfterCopy, statusErr = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{
+			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
+		}
+
 		return nil
 	})
 
@@ -223,10 +241,24 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 		}
 	}
 
-	// mount the volume
-	err = vs.store.MountVolume(needle.VolumeId(req.VolumeId))
+	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 {
+		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), func(targetVolume *storage.Volume) error {
+		if !shouldValidateCopyCounts {
+			return nil
+		}
+		return checkCopyCounts(sourceVolumeStatusAfterCopy, targetVolume.FileCount(), targetVolume.DeletedCount())
+	})
 	if err != nil {
-		return fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
+		return fmt.Errorf("failed to mount or validate volume %d: %w", req.VolumeId, err)
 	}
 
 	if err = stream.Send(&volume_server_pb.VolumeCopyResponse{
@@ -266,11 +298,8 @@ func (vs *VolumeServer) doCopyFileWithThrottler(client volume_server_pb.VolumeSe
 
 }
 
-/*
-*
-only check the differ of the file size
-todo: maybe should check the received count and deleted count of the volume
-*/
+// checkCopyFiles verifies the copied file sizes. Record counts are checked
+// after the target volume is mounted, when the target needle map is available.
 func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse, hasRemoteDatFile bool, idxFileName, datFileName string) error {
 	stat, err := os.Stat(idxFileName)
 	if err != nil {
@@ -300,6 +329,22 @@ func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse
 	return nil
 }
 
+func checkCopyCounts(origin *volume_server_pb.VolumeStatusResponse, targetFileCount, targetDeletedCount uint64) error {
+	if origin.FileCount != targetFileCount {
+		return fmt.Errorf("target file count [%d] is not same as origin file count [%d]", targetFileCount, origin.FileCount)
+	}
+	if origin.FileDeletedCount != targetDeletedCount {
+		return fmt.Errorf("target deleted count [%d] is not same as origin deleted count [%d]", targetDeletedCount, origin.FileDeletedCount)
+	}
+	return nil
+}
+
+func copyCountsStable(before, after *volume_server_pb.VolumeStatusResponse) bool {
+	return before != nil && after != nil &&
+		before.FileCount == after.FileCount &&
+		before.FileDeletedCount == after.FileDeletedCount
+}
+
 func findLastAppendAtNsFromCopiedFiles(idxFileName, datFileName string, version needle.Version) (uint64, error) {
 	if version < needle.Version3 {
 		return 0, nil

weed/server/volume_grpc_copy_count_test.go

@@ -0,0 +1,148 @@
+package weed_server
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+)
+
+func TestCheckCopyCounts(t *testing.T) {
+	tests := []struct {
+		name              string
+		sourceFileCount   uint64
+		sourceDeleteCount uint64
+		targetFileCount   uint64
+		targetDeleteCount uint64
+		wantErr           string
+	}{
+		{
+			name: "empty volume counts match",
+		},
+		{
+			name:              "file and deleted counts match",
+			sourceFileCount:   10,
+			sourceDeleteCount: 3,
+			targetFileCount:   10,
+			targetDeleteCount: 3,
+		},
+		{
+			name:              "file count differs",
+			sourceFileCount:   10,
+			sourceDeleteCount: 3,
+			targetFileCount:   9,
+			targetDeleteCount: 3,
+			wantErr:           "file count",
+		},
+		{
+			name:              "deleted count differs",
+			sourceFileCount:   10,
+			sourceDeleteCount: 3,
+			targetFileCount:   10,
+			targetDeleteCount: 2,
+			wantErr:           "deleted count",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := checkCopyCounts(
+				&volume_server_pb.VolumeStatusResponse{
+					FileCount:        tt.sourceFileCount,
+					FileDeletedCount: tt.sourceDeleteCount,
+				},
+				tt.targetFileCount,
+				tt.targetDeleteCount,
+			)
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("checkCopyCounts() error = %v", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("checkCopyCounts() error = %v, want substring %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestCopyCountsStable(t *testing.T) {
+	tests := []struct {
+		name   string
+		before *volume_server_pb.VolumeStatusResponse
+		after  *volume_server_pb.VolumeStatusResponse
+		want   bool
+	}{
+		{
+			name:   "empty volume is stable",
+			before: &volume_server_pb.VolumeStatusResponse{},
+			after:  &volume_server_pb.VolumeStatusResponse{},
+			want:   true,
+		},
+		{
+			name: "matching file and deleted counts are stable",
+			before: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+			},
+			after: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+			},
+			want: true,
+		},
+		{
+			name: "read-only state changes without count changes are stable",
+			before: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+				IsReadOnly:       true,
+			},
+			after: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+				IsReadOnly:       false,
+			},
+			want: true,
+		},
+		{
+			name: "changed file count is not stable",
+			before: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+			},
+			after: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        9,
+				FileDeletedCount: 3,
+			},
+		},
+		{
+			name: "changed deleted count is not stable",
+			before: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 3,
+			},
+			after: &volume_server_pb.VolumeStatusResponse{
+				FileCount:        10,
+				FileDeletedCount: 2,
+			},
+		},
+		{
+			name:  "missing before status is not stable",
+			after: &volume_server_pb.VolumeStatusResponse{},
+		},
+		{
+			name:   "missing after status is not stable",
+			before: &volume_server_pb.VolumeStatusResponse{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := copyCountsStable(tt.before, tt.after); got != tt.want {
+				t.Fatalf("copyCountsStable() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

weed/storage/store.go

@@ -967,11 +967,30 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 }
 
 func (s *Store) MountVolume(i needle.VolumeId) error {
+	return s.mountVolume(i, nil)
+}
+
+// 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, validator func(*Volume) error) error {
+	return s.mountVolume(i, validator)
+}
+
+func (s *Store) mountVolume(i needle.VolumeId, validator func(*Volume) error) error {
 	for diskId, location := range s.Locations {
 		if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind); 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 {
+				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)
+					}
+					return err
+				}
+			}
 			readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
 			s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
 				Id:                uint32(v.Id),

weed/storage/store_mount_validation_test.go

@@ -0,0 +1,48 @@
+package storage
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) {
+	dir := t.TempDir()
+	const vid = needle.VolumeId(17)
+
+	volume, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory,
+		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	volume.Close()
+
+	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+		[]string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "",
+		NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3,
+		stats.DefaultDiskIOProbeConfig())
+	t.Cleanup(store.Close)
+
+	validationErr := errors.New("copy counts differ")
+	err = store.MountVolumeWithValidator(vid, func(*Volume) error {
+		select {
+		case <-store.NewVolumesChan:
+			t.Fatal("volume was announced before validation completed")
+		default:
+		}
+		return validationErr
+	})
+
+	require.ErrorIs(t, err, validationErr)
+	require.Nil(t, store.GetVolume(vid))
+	select {
+	case message := <-store.NewVolumesChan:
+		t.Fatalf("volume was announced after validation failed: %+v", message)
+	default:
+	}
+}