fix(storage): never let an empty .dat delete healthy distributed EC shards (#9930)

* fix(storage): never let an empty .dat delete healthy distributed EC shards

A leftover empty .dat stub (a phantom from the pre-fix loader; zero
needles) next to a distributed EC volume's local shards made startup
classify the volume as an interrupted local encode: validateEcVolume
requires >= dataShards local shards when a .dat is present, fails with
the 1-2 shards a distributed volume keeps per disk, and the cleanup
deletes those shards -- the only copies of that part of the volume.
Repeated across restart waves this destroys enough shards cluster-wide
to make the volume unrecoverable.

Go:
- loadExistingVolume: hoist the empty-stub sweep above the EC presence
  checks. Previously the .vif-next-to-.ecx guard returned before the
  sweep ever ran, so exactly the dangerous layout (stub + .ecx + local
  shards) kept its stub and then lost its shards in loadAllEcShards.
- validateEcVolume / checkDatFileExists: treat a .dat <= a superblock
  (zero needles) as absent. An empty .dat cannot be the encode source,
  so it must never gate shard deletion; this also covers stubs without
  a .vif, which the sweep cannot prove are EC leftovers.

Rust mirror (seaweed-volume): the same gate in validate_ec_volume and
check_dat_file_exists (the Rust sweep already ran before validation);
the volume-load skip keeps a plain existence check so fresh,
needle-less volumes still load.

Regression tests in Go and Rust reproduce the production layout (a
zero-byte .dat beside .ecx/.ecj and two shards of a 10+4 volume, with
and without a .vif) and fail without the fix with the shards deleted.

* fix(ec): gate source volume deletion on a recoverable shard set

After EC encode, the shell command and the (plugin) worker task refused
to delete the source volume unless every shard was present, and aborted
otherwise -- leaving the source .dat next to live shards, exactly the
mixed state the startup cleanup mishandles.

Replace the full-set requirement with a recoverability gate shared by
both callers (RequireRecoverableShardSet): deleting a non-empty source
.dat requires at least dataShards distinct shards cluster-wide. Below
that the source is kept and the encode fails as before. A degraded but
recoverable set (>= dataShards, < total) now proceeds with a warning
instead of aborting: the missing shards can be rebuilt from the
survivors, while keeping the source would preserve the dangerous mixed
state. Empty stub replicas are still swept unguarded (OnlyEmpty) -- an
empty .dat has nothing to lose.

dataShards/totalShards stay parameters so enterprise custom EC ratios
share the helper verbatim.

* test(ec): use recoverable shard verification gate

weed/shell/command_ec_encode.go

@@ -594,12 +594,16 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum
 	// volume-server heartbeats, so freshly distributed shards may not all be
 	// visible in the master topology immediately. Poll a few times before
 	// concluding the shard set is incomplete, so a heartbeat-propagation lag is
-	// not mistaken for missing data (which would abort the encode). Genuine loss
-	// still fails after the retries are exhausted.
+	// not mistaken for missing data. After the retries: a volume below the
+	// recoverable threshold (dataShards) aborts the deletion; a recoverable
+	// but degraded set proceeds with a warning, since the missing shards can
+	// be rebuilt from the survivors while keeping the source next to live
+	// shards is the more dangerous mixed state.
 	const maxAttempts = 10
 	const retryInterval = 2 * time.Second
 
 	var lastErr error
+	var lastDegraded []string
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		topoInfo, _, err := collectTopologyInfo(commandEnv, 0)
 		if err != nil {
@@ -607,6 +611,7 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum
 		}
 
 		lastErr = nil
+		lastDegraded = lastDegraded[:0]
 		for _, vid := range volumeIds {
 			nodeShards, _ := collectEcNodeShardsInfo(topoInfo, vid, diskType)
 
@@ -616,7 +621,8 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum
 			}
 
 			totalShards := erasure_coding.TotalShardsCount
-			if err := erasure_coding.RequireFullShardSet(uint32(vid), union, totalShards); err != nil {
+			degraded, err := erasure_coding.RequireRecoverableShardSet(uint32(vid), union, erasure_coding.DataShardsCount, totalShards)
+			if err != nil {
 				summary := make([]string, 0, len(nodeShards))
 				for node, info := range nodeShards {
 					summary = append(summary, fmt.Sprintf("%s=%s", node, info.String()))
@@ -625,23 +631,32 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum
 				lastErr = fmt.Errorf("volume %d: %w (observed: %v)", vid, err, summary)
 				break
 			}
+			if degraded {
+				lastDegraded = append(lastDegraded, fmt.Sprintf("volume %d: %d/%d shards", vid, union.Count(), totalShards))
+				continue
+			}
 
 			glog.V(0).Infof("EC shard verification ok for volume %d on diskType %q: %d/%d shards present across %d nodes",
 				vid, diskType.ReadableString(), union.Count(), totalShards, len(nodeShards))
 		}
 
-		if lastErr == nil {
+		if lastErr == nil && len(lastDegraded) == 0 {
 			return nil
 		}
 		if attempt < maxAttempts-1 {
-			glog.V(0).Infof("EC shard verification incomplete (attempt %d/%d), waiting for shard locations to propagate: %v",
-				attempt+1, maxAttempts, lastErr)
+			glog.V(0).Infof("EC shard verification incomplete (attempt %d/%d), waiting for shard locations to propagate: %v %v",
+				attempt+1, maxAttempts, lastErr, lastDegraded)
 			time.Sleep(retryInterval)
 		}
 	}
 
-	glog.Errorf("EC shard verification failed after %d attempts: %v", maxAttempts, lastErr)
-	return lastErr
+	if lastErr != nil {
+		glog.Errorf("EC shard verification failed after %d attempts: %v", maxAttempts, lastErr)
+		return lastErr
+	}
+	glog.Warningf("EC shard set incomplete but recoverable after %d attempts, proceeding with source deletion (rebuild missing shards with ec.rebuild): %v",
+		maxAttempts, lastDegraded)
+	return nil
 }
 
 // doDeleteVolumesWithLocations deletes volumes using pre-collected location information