refactor: share volume and EC shard move logic between shell and workers (#10727)

* operation: add shared volume_move package for volume and EC shard moves

The shell commands (volume.move, volume.balance, ec.balance, tier moves)
and the maintenance workers (balance, ec_balance) each carried their own
copy of the move RPC sequences, and the copies had drifted: the worker
verified the target before deleting the source but dropped the disk
type and IO throttle; the shell passed those but deleted the source
unverified.

volume_move.Mover carries the merged sequences, keeping the stricter
behavior from each side:

- LiveMoveVolume: check-then-hard-freeze the source (VolumeStatus's
  IsReadOnly also covers low-disk and readonly-but-can-delete states,
  which still accept needle deletes), copy with disk type and IO
  throttle, tail, verify the target is not behind the source before the
  destructive source delete (a target that is ahead holds writes it
  accepted during the tail and the move commits to keep them), and
  restore the source's writability when a failure precedes the delete
  and this move did the freezing. Aborts clean up the incomplete target
  copy; a failed cleanup or an ambiguous source delete keeps the source
  readonly (ErrSourceKeptReadonly) so callers do not thaw a source next
  to a possibly-authoritative copy. With a readonly source, an existing
  or unknown-state target refuses the move outright: no client-side
  observation can prove such a copy is a stale remnant rather than the
  authoritative copy of an unfinished move.
- MoveEcShards: copy with the .ecx/.ecj/.vif/.ecsum sidecars, mount,
  verify the target registered every shard before unmount+delete on the
  source, and reject same-server moves (the EC delete is server-wide).

Server identity is the grpc endpoint (SameServer), so node:8080 and
node:8080.18080 compare equal while test servers sharing a degenerate
HTTP address stay distinct; addresses are validated non-fatally before
dialing and before being embedded in copy/tail requests, since both the
client dialer and the receiving server normalize them through a parser
that aborts the process on a malformed port. The Rust volume server's
codes.NotFound counts as a definitively absent probe answer alongside
the Go server's plain-error code Unknown.

All RPCs go through an injectable ClientFunc, so the sequences are unit
tested against a fake volume server client: RPC order, request fields,
and that verification failures keep the source intact.

* shell, worker: delegate volume and EC shard moves to operation/volume_move

LiveMoveVolume and the copy/tail/delete/mark-writable helpers become
thin wrappers over the shared mover, keeping their signatures; the EC
helpers keep their per-step output and delegate the RPCs. BalanceTask
and ECBalanceTask keep their parameter validation, progress reporting,
and guards (same-node cross-disk rejection, dedup keep-node
verification, shard ids range-checked before the uint8 narrowing) and
hand the RPC sequences to the mover. volume.tier.move skips its
thaw-on-failure when the mover deliberately kept the source readonly,
since reopening the replicas beside a possibly-authoritative target
copy would fork the volume.

The tail-failure tolerance moves inside the mover: a failed tail is
tolerated only when the volume was already readonly before the move
began, backstopped by a stability re-read across the idle window, so
volume.balance's -skipTailError-by-readonly heuristic and tier-move's
unconditional skip both become the same authoritative rule.

* volume_move: keep the source readonly when a failed copy leaves a target of unknown origin

A failed copy can leave a complete, mounted copy on the target (the
server finishes after the client loses the stream). The abort probed
the target only when its pre-copy state was known-absent; an unknown
prior state skipped both the probe and the cleanup and then reopened
the source - two writable replicas of one volume, diverging from the
next write on.

The abort now probes the target on every failed copy and restores the
source only when the target provably holds nothing. A copy whose
provenance cannot be proven (unknown prior state, a pre-existing
replica, or an unreachable target) is never deleted, and the source
stays readonly with ErrSourceKeptReadonly naming the recovery.

* test: teach the plugin worker harness the shared move sequence

The fake volume server lacked VolumeStatus, which the shared mover now
issues before freezing the source, and the batch execution test's
status-read accounting predates the pre-copy target probe and the
verification reads. Mirrors the harness the enterprise tree already
carries.

weed/shell/command_ec_common.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"regexp"
 	"slices"
 	"sort"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/operation/volume_move"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
@@ -389,48 +391,19 @@ func oneServerCopyAndMountEcShardsFromSource(grpcDialOption grpc.DialOption,
 	fmt.Printf("allocate %d.%v %s => %s\n", volumeId, shardIdsToCopy, existingLocation, targetServer.info.Id)
 
 	targetAddress := pb.NewServerAddressFromDataNode(targetServer.info)
-	err = operation.WithVolumeServerClient(false, targetAddress, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-
-		if targetAddress != existingLocation {
-			fmt.Printf("copy %d.%v %s => %s\n", volumeId, shardIdsToCopy, existingLocation, targetServer.info.Id)
-			_, copyErr := volumeServerClient.VolumeEcShardsCopy(context.Background(), &volume_server_pb.VolumeEcShardsCopyRequest{
-				VolumeId:       uint32(volumeId),
-				Collection:     collection,
-				ShardIds:       erasure_coding.ShardIdsToUint32(shardIdsToCopy),
-				CopyEcxFile:    true,
-				CopyEcjFile:    true,
-				CopyVifFile:    true,
-				CopyEcsumFile:  true, // propagate the bitrot sidecar with the shards (no-op if the source has none)
-				SourceDataNode: string(existingLocation),
-				DiskId:         destDiskId,
-			})
-			if copyErr != nil {
-				return fmt.Errorf("copy %d.%v %s => %s : %v\n", volumeId, shardIdsToCopy, existingLocation, targetServer.info.Id, copyErr)
-			}
-		}
-
-		fmt.Printf("mount %d.%v on %s\n", volumeId, shardIdsToCopy, targetServer.info.Id)
-		_, mountErr := volumeServerClient.VolumeEcShardsMount(context.Background(), &volume_server_pb.VolumeEcShardsMountRequest{
-			VolumeId:   uint32(volumeId),
-			Collection: collection,
-			ShardIds:   erasure_coding.ShardIdsToUint32(shardIdsToCopy),
-		})
-		if mountErr != nil {
-			return fmt.Errorf("mount %d.%v on %s : %v\n", volumeId, shardIdsToCopy, targetServer.info.Id, mountErr)
-		}
-
-		if targetAddress != existingLocation {
-			copiedShardIds = shardIdsToCopy
-			glog.V(0).Infof("%s ec volume %d deletes shards %+v", existingLocation, volumeId, copiedShardIds)
-		}
-
-		return nil
-	})
-
+	err = volume_move.NewMover(grpcDialOption).CopyAndMountEcShards(context.Background(), volumeId, collection, shardIdsToCopy, existingLocation, targetAddress, destDiskId, os.Stdout)
 	if err != nil {
 		return
 	}
 
+	// SameServer, not ==: a representation mismatch here would report a
+	// same-server mount-in-place as a copy and have the caller delete the
+	// shards it kept.
+	if !volume_move.SameServer(targetAddress, existingLocation) {
+		copiedShardIds = shardIdsToCopy
+		glog.V(0).Infof("%s ec volume %d deletes shards %+v", existingLocation, volumeId, copiedShardIds)
+	}
+
 	return
 }
 
@@ -582,15 +555,7 @@ func sourceServerDeleteEcShards(grpcDialOption grpc.DialOption, collection strin
 
 	fmt.Printf("delete %d.%v from %s\n", volumeId, toBeDeletedShardIds, sourceLocation)
 
-	return operation.WithVolumeServerClient(false, sourceLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-		_, deleteErr := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{
-			VolumeId:   uint32(volumeId),
-			Collection: collection,
-			ShardIds:   erasure_coding.ShardIdsToUint32(toBeDeletedShardIds),
-		})
-		return deleteErr
-	})
-
+	return volume_move.NewMover(grpcDialOption).DeleteEcShards(context.Background(), volumeId, collection, sourceLocation, toBeDeletedShardIds)
 }
 
 // errFullTeardownNotAcked marks a reachable server that completed the delete RPC
@@ -649,27 +614,14 @@ func unmountEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, s
 
 	fmt.Printf("unmount %d.%v from %s\n", volumeId, toBeUnmountedShardIds, sourceLocation)
 
-	return operation.WithVolumeServerClient(false, sourceLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-		_, deleteErr := volumeServerClient.VolumeEcShardsUnmount(context.Background(), &volume_server_pb.VolumeEcShardsUnmountRequest{
-			VolumeId: uint32(volumeId),
-			ShardIds: erasure_coding.ShardIdsToUint32(toBeUnmountedShardIds),
-		})
-		return deleteErr
-	})
+	return volume_move.NewMover(grpcDialOption).UnmountEcShards(context.Background(), volumeId, sourceLocation, toBeUnmountedShardIds)
 }
 
 func mountEcShards(grpcDialOption grpc.DialOption, collection string, volumeId needle.VolumeId, sourceLocation pb.ServerAddress, toBeMountedShardIds []erasure_coding.ShardId) error {
 
 	fmt.Printf("mount %d.%v on %s\n", volumeId, toBeMountedShardIds, sourceLocation)
 
-	return operation.WithVolumeServerClient(false, sourceLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-		_, mountErr := volumeServerClient.VolumeEcShardsMount(context.Background(), &volume_server_pb.VolumeEcShardsMountRequest{
-			VolumeId:   uint32(volumeId),
-			Collection: collection,
-			ShardIds:   erasure_coding.ShardIdsToUint32(toBeMountedShardIds),
-		})
-		return mountErr
-	})
+	return volume_move.NewMover(grpcDialOption).MountEcShards(context.Background(), volumeId, collection, sourceLocation, toBeMountedShardIds)
 }
 
 func ceilDivide(a, b int) int {
@@ -1157,23 +1109,26 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e
 	return ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk)
 }
 
-// applyShardMoveRPC copies a shard to the destination disk, then unmounts and
-// deletes it on the source. It does not touch the in-memory model, so it is safe
-// to run concurrently across the moves of a phase.
+// applyShardMoveRPC copies a shard to the destination disk, verifies the
+// destination registered it, then unmounts and deletes it on the source. It
+// does not touch the in-memory model, so it is safe to run concurrently across
+// the moves of a phase.
 func (ecb *ecBalancer) applyShardMoveRPC(src, dst *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, destDiskId uint32) error {
-	grpcDialOption := ecb.commandEnv.option.GrpcDialOption
 	srcAddr := pb.NewServerAddressFromDataNode(src.info)
-	copiedShardIds, err := oneServerCopyAndMountEcShardsFromSource(grpcDialOption, dst, []erasure_coding.ShardId{shardId}, vid, collection, srcAddr, destDiskId)
-	if err != nil {
-		return err
-	}
-	if len(copiedShardIds) == 0 {
+	dstAddr := pb.NewServerAddressFromDataNode(dst.info)
+	if volume_move.SameServer(srcAddr, dstAddr) {
+		// A same-server (cross-disk) move cannot be expressed with these RPCs;
+		// leave the shard where it is.
 		return nil
 	}
-	if err := unmountEcShards(grpcDialOption, vid, srcAddr, copiedShardIds); err != nil {
-		return err
-	}
-	return sourceServerDeleteEcShards(grpcDialOption, collection, vid, srcAddr, copiedShardIds)
+	return volume_move.NewMover(ecb.commandEnv.option.GrpcDialOption).MoveEcShards(context.Background(), volume_move.EcShardMove{
+		VolumeId:   vid,
+		Collection: collection,
+		ShardIds:   []erasure_coding.ShardId{shardId},
+		Source:     srcAddr,
+		Target:     dstAddr,
+		TargetDisk: destDiskId,
+	}, volume_move.EcMoveOptions{Writer: os.Stdout})
 }
 
 // parseVolumeIdsFlag parses a comma-separated -volumeIds flag value, dropping