fix(replication): verify-before-destroy in VolumeCopy, check.disk, and over-replication trim (#9943)

* volume: verify before destroy in VolumeCopy and replication repair

Four data-safety fixes around copy/repair paths that could destroy or
resurrect data before verifying the source or survivors.

(a) VolumeCopy no longer deletes a pre-existing local replica up front.
The delete is deferred until ReadVolumeFileStatus on the source succeeds,
so a transient source outage (or a retry after one) can no longer wipe a
healthy destination replica. Gated on source readability only; size/count
comparisons are intentionally not used because they invert legitimately
after divergent vacuum/compaction. Mirrored in the Rust volume server.

(b) volume.check.disk no longer resurrects vacuumed-deleted needles. A
key present-and-live on the source but entirely absent on the target is
ambiguous: it may be a genuine missing write, or a needle deleted on the
target and then vacuumed (its index entry and any tombstone are gone). An
individual needle AppendAtNs has no monotonic relation to a vacuum
watermark, so the old cutoff heuristic could not tell them apart. Without
positive proof the absence is a missing write, the safe default is to NOT
push it back. Tradeoff: a real missing write may go unrepaired until a
tombstone-aware path exists, but we never raise back deleted data.

(c) Over-replication trim no longer resurrects needles or removes the
wrong replica. The pre-delete sync now runs read-only (divergence check
only) instead of writing the doomed replica's needles into the survivor.
pickOneReplicaToDelete only ever removes the smallest of multiple healthy
writable replicas; it refuses the trim when doing so would leave only
read-only/integrity-flagged survivors, since file_count>0 alone cannot
prove the survivor's .dat is readable.

(d) Incomplete-volume (.note) cleanup keeps the shared .vif when an .ecx
for the same vid coexists on the disk, so removing an interrupted regular
copy cannot strip a coexisting EC volume's info file. VolumeCopy now
surfaces .note write/remove errors instead of ignoring them. In the Rust
volume server (where a persisting note is actually reachable) the .note
check moves below the empty-stub sweep and EC validation, keeps the .vif
on EC coexistence, and the mount path fails when a .note still persists.

* shell: scope the over-replication writable-survivor guard to the trim path only

The writable-survivor guard (never trim down to a read-only survivor) lived
inside the shared pickOneReplicaToDelete, so it also gated the misplaced-volume
relocation via pickOneMisplacedVolume -- a misplaced read-only volume (e.g. a
full one) would silently stop being rebalanced. Extract pickSmallestReplica
for the relocation path (which deletes-and-recreates and must act on read-only
replicas), and keep the writable-survivor guard only in pickOneReplicaToDelete
used by the over-replication trim.

* seaweed-volume: recompute keep_vif after invalid-EC cleanup in the .note path

keep_vif used the pre-validation ecx_exists snapshot, so when the EC-validation
step above removed the invalid .ecx/shards, the .note cleanup still preserved a
now-orphaned .vif. Re-check .ecx existence at cleanup time, matching the Go
hasEcxFile re-check.

* shell: keep placement when picking an over-replication victim to delete

The trim picked the smallest writable replica without regard to placement, so
it could delete the only replica in a required failure domain (e.g. with "100"
and replicas dc1 + two in dc2, deleting dc1 leaves both survivors in dc2).
Prefer a writable replica whose removal still satisfies placement, falling back
to the smallest writable only when none does.

weed/shell/command_volume_fix_replication.go

@@ -219,7 +219,12 @@ func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[ui
 
 type SelectOneVolumeFunc func(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica
 
-func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, commandEnv *CommandEnv) (err error) {
+// checkOneVolume compares the index of replica a against b. With
+// applyChanges=false it is a read-only divergence check; the over-replication
+// trim must use that mode so it does not push the soon-to-be-deleted replica's
+// needles into the survivor (which would resurrect data and is the opposite of
+// a safe trim).
+func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, commandEnv *CommandEnv, applyChanges bool) (err error) {
 	aDB, bDB := needle_map.NewMemDb(), needle_map.NewMemDb()
 	defer func() {
 		aDB.Close()
@@ -232,7 +237,7 @@ func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, comman
 		now:        time.Now(),
 
 		verbose:            false,
-		applyChanges:       true,
+		applyChanges:       applyChanges,
 		syncDeletions:      false,
 		nonRepairThreshold: float64(1),
 	}
@@ -261,6 +266,10 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr
 		replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replicas[0].info.ReplicaPlacement))
 
 		replica := selectOneVolumeFn(replicas, replicaPlacement)
+		if replica == nil {
+			fmt.Fprintf(writer, "skip trimming volume %d: no safe replica to delete (would leave only read-only survivors)\n", vid)
+			continue
+		}
 
 		// check collection name pattern
 		if *c.collectionPattern != "" {
@@ -302,7 +311,9 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr
 				if replicaB.location.dataNode == replica.location.dataNode {
 					continue
 				}
-				if checkErr = checkOneVolume(replica, replicaB, writer, commandEnv); checkErr != nil {
+				// Read-only divergence check only: never write the doomed
+				// replica's needles into a survivor while trimming.
+				if checkErr = checkOneVolume(replica, replicaB, writer, commandEnv, false); checkErr != nil {
 					fmt.Fprintf(writer, "sync volume %d on %s and %s: %v\n", replica.info.Id, replica.location.dataNode.Id, replicaB.location.dataNode.Id, checkErr)
 					break
 				}
@@ -625,8 +636,20 @@ func countReplicas(replicas []*VolumeReplica) (diffDc, diffRack, diffNode map[st
 	return
 }
 
-func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica {
-	slices.SortFunc(replicas, func(a, b *VolumeReplica) int {
+// pickOneReplicaToDelete selects the replica to trim when over-replicated.
+// It only ever removes the smallest of multiple healthy writable replicas: a
+// ReadOnly/integrity-flagged replica is never chosen for deletion, and the
+// trim is refused (returns nil) when removing a writable replica would leave
+// only ReadOnly survivors. VolumeStatus file_count>0 alone cannot prove the
+// survivors' .dat is readable, so we do not over-claim survivor health.
+// pickSmallestReplica returns the smallest replica (ties broken by oldest then
+// lowest compact revision), or nil for an empty set.
+func pickSmallestReplica(replicas []*VolumeReplica) *VolumeReplica {
+	if len(replicas) == 0 {
+		return nil
+	}
+	sorted := slices.Clone(replicas)
+	slices.SortFunc(sorted, func(a, b *VolumeReplica) int {
 		if a.info.Size != b.info.Size {
 			return int(a.info.Size - b.info.Size)
 		}
@@ -638,9 +661,39 @@ func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_b
 		}
 		return 0
 	})
+	return sorted[0]
+}
 
-	return replicas[0]
-
+func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica {
+	// Over-replication trim: only ever remove a writable replica, and only
+	// when another writable one survives, so a healthy copy is never deleted
+	// down to a read-only (e.g. full or integrity-flagged) survivor.
+	var writable []*VolumeReplica
+	for _, r := range replicas {
+		if !r.info.ReadOnly {
+			writable = append(writable, r)
+		}
+	}
+	if len(writable) < 2 {
+		return nil
+	}
+	// Prefer a writable replica whose removal still satisfies placement, so the
+	// trim does not strip the only replica in a required failure domain. Fall
+	// back to the smallest writable if none keeps placement (a later misplaced
+	// cycle then re-balances).
+	var placementSafe []*VolumeReplica
+	for i, r := range replicas {
+		if r.info.ReadOnly {
+			continue
+		}
+		if !isMisplaced(otherThan(replicas, i), replicaPlacement) {
+			placementSafe = append(placementSafe, r)
+		}
+	}
+	if len(placementSafe) > 0 {
+		return pickSmallestReplica(placementSafe)
+	}
+	return pickSmallestReplica(writable)
 }
 
 // check and fix misplaced volumes
@@ -669,6 +722,10 @@ func otherThan(replicas []*VolumeReplica, index int) (others []*VolumeReplica) {
 
 func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) (toDelete *VolumeReplica) {
 
+	// Relocation, not over-replication: pick the smallest replica to delete
+	// and recreate at a correct placement. Unlike the trim this must still act
+	// on read-only replicas (e.g. a full but misplaced volume), so it does not
+	// use pickOneReplicaToDelete's writable-survivor guard.
 	var deletionCandidates []*VolumeReplica
 	for i := 0; i < len(replicas); i++ {
 		others := otherThan(replicas, i)
@@ -676,10 +733,10 @@ func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_b
 			deletionCandidates = append(deletionCandidates, replicas[i])
 		}
 	}
-	if len(deletionCandidates) > 0 {
-		return pickOneReplicaToDelete(deletionCandidates, replicaPlacement)
+	if toDelete = pickSmallestReplica(deletionCandidates); toDelete != nil {
+		return toDelete
 	}
 
-	return pickOneReplicaToDelete(replicas, replicaPlacement)
+	return pickSmallestReplica(replicas)
 
 }