ec: pin auto-selected shard placement to the disk that already owns the shard (#11029)

* ec: pin auto-selected shard placement to the disk that already owns the shard

A multi-disk server legitimately mounts one EC volume on several disks, so
FindEcShardTargetLocation's per-volume tiers tie at "mounted" and the
free-shard-count tie-break decides — pointing at whichever disk is emptier,
not at the disk that already holds the shard being placed. A re-copy of a
shard the server already has (a retried ec.balance / ec.rebuild move) then
lands on a sibling disk, and both disks register the same (volume, shard id):
the shard is reported to the master from two disk ids, and which claimant
serves reads or survives a later unmount/delete becomes an accident of
Locations order.

Add a tier above "mounted": a disk that already claims one of the shard ids
being placed wins, ahead of the space filters too — re-copying in place
needs no new shard slot, and a genuinely full disk should fail the write
rather than silently split the claim. Applied to the Go selector and the
VolumeEcShardsCopy auto-select (ReceiveFile refuses mounted EC volumes, so
no claim can exist there) and mirrored in the Rust volume server.

Claude-Session: https://claude.ai/code/session_01AWpefvdi4U3HLng18x5CJ9

* ec: refuse a copy batch whose shards are already owned by different disks

Review follow-up: ownership-aware selection ranks a mixed-owner batch
(shard 0 on disk A, shard 2 on disk B — the legitimate multi-disk spread)
into one destination, so the copy would still duplicate the losing disk's
claim. No production caller sends such a batch (balance moves one shard,
rebuild and encode copy shards the target lacks), so fail closed: report
every owning disk via Store.EcShardOwnerDisks and refuse the copy with an
error naming them, telling the caller to split per shard or pass disk_id.
Go and Rust, with unit tests for the owner-reporting contract.

Claude-Session: https://claude.ai/code/session_01AWpefvdi4U3HLng18x5CJ9

weed/server/volume_grpc_erasure_coding.go

@@ -343,15 +343,33 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 		location = vs.store.Locations[req.DiskId]
 		glog.V(1).Infof("Using disk %d for EC shard copy: %s", req.DiskId, location.Directory)
 	} else {
-		// Auto-select the target disk: prefer a disk that already has the
-		// EC volume mounted, then a disk that owns the .ecx on disk (the
-		// volume hasn't been mounted yet — relevant for ec.rebuild, where
-		// only the first shard carries .ecx and subsequent shards must
-		// land on the same disk; see #9212), then any HDD, then any disk.
-		// Pass the build's default data-shard count for free-slot maths;
-		// the helper takes it as a parameter so custom-ratio builds (e.g.
-		// enterprise) can swap it without touching this file.
-		location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount)
+		// Auto-select the target disk: prefer a disk that already owns one
+		// of the shards being copied (a retried move must overwrite in
+		// place, not leave two disks of this server claiming the same
+		// shard), then a disk that already has the EC volume mounted, then
+		// a disk that owns the .ecx on disk (the volume hasn't been mounted
+		// yet — relevant for ec.rebuild, where only the first shard carries
+		// .ecx and subsequent shards must land on the same disk; see
+		// #9212), then any HDD, then any disk. Pass the build's default
+		// data-shard count for free-slot maths; the helper takes it as a
+		// parameter so custom-ratio builds (e.g. enterprise) can swap it
+		// without touching this file.
+		shardIds := make([]erasure_coding.ShardId, 0, len(req.ShardIds))
+		for _, shardId := range req.ShardIds {
+			shardIds = append(shardIds, erasure_coding.ShardId(shardId))
+		}
+		// A batch whose requested shards are already owned by different local
+		// disks has no single correct destination: writing them all to one
+		// disk would duplicate the other disks' claims. Refuse so the caller
+		// splits the batch per shard (or chooses explicitly via disk_id).
+		if owners := vs.store.EcShardOwnerDisks(needle.VolumeId(req.VolumeId), shardIds); len(owners) > 1 {
+			dirs := make([]string, 0, len(owners))
+			for _, owner := range owners {
+				dirs = append(dirs, owner.Directory)
+			}
+			return nil, fmt.Errorf("volume %d shards %v are already owned by multiple local disks %v: no single destination; copy per shard or pass disk_id", req.VolumeId, req.ShardIds, dirs)
+		}
+		location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount, shardIds...)
 		if location == nil {
 			return nil, fmt.Errorf("no space left")
 		}