fix(erasure_coding): surface replica delete failures from EC task (#9184) (#9187)

* test(erasure_coding): reproduce #9184 deleteOriginalVolume swallowing errors

ErasureCodingTask.deleteOriginalVolume logs a warning when any replica
VolumeDelete fails and then returns nil, so the EC task reports
success to the admin even when a source replica survives. That stale
replica lets a later detection scan re-propose the same volume and,
once retried, drives the mounted-shard-truncation corruption that
issue 9184 also describes.

Reproducer: wire one reachable replica (succeeds) and one unreachable
replica (fails) and assert the function currently returns nil. After
the fix the function must surface the replica failure so the task is
retried rather than marked done, and this test needs to be inverted.

* fix(erasure_coding): surface replica delete failures from EC task

ErasureCodingTask.deleteOriginalVolume previously logged a warning
and returned nil when any VolumeDelete against a source replica
failed. The EC task therefore reported overall success to the admin
even when a source replica stayed on disk, which let a later
detection scan propose a duplicate EC encoding of the same volume.
The retry then walked the ReceiveFile path against servers that
already had mounted EC shards for the volume, truncating the live
shard files in place (the other half of #9184).

This change returns an error describing the per-replica failures
after the best-effort delete pass, so the task is marked failed
instead of silently moving on. Successful deletes are still applied
(per-replica progress is preserved); only the final return changes.

When combined with the ReceiveFile mount-safety check, a stuck
original replica now produces loud, actionable failures instead of
silent corruption.

Tests:
- TestDeleteOriginalVolumeSurfacesReplicaFailures: asserts an error
  is returned and names the unreachable replica, while the reachable
  replica still gets deleted.
- TestDeleteOriginalVolumeSucceedsWhenAllReplicasReachable: pins the
  happy path.

weed/worker/tasks/erasure_coding/ec_task.go

@@ -596,7 +596,6 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error {
 		}
 	}
 
-	// Report results
 	if len(deleteErrors) > 0 {
 		t.GetLogger().WithFields(map[string]interface{}{
 			"volume_id":      t.volumeID,
@@ -605,16 +604,19 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error {
 			"total_replicas": len(replicas),
 			"success_rate":   float64(successCount) / float64(len(replicas)) * 100,
 			"errors":         deleteErrors,
-		}).Warning("Some volume deletions failed")
-		// Don't return error - EC task should still be considered successful if shards are mounted
-	} else {
-		t.GetLogger().WithFields(map[string]interface{}{
-			"volume_id":       t.volumeID,
-			"replica_count":   len(replicas),
-			"replica_servers": replicas,
-		}).Info("Successfully deleted volume from all replica servers")
+		}).Error("Failed to delete some original volume replicas after EC encoding")
+		// A surviving source replica lets a later detection scan re-propose
+		// EC on the same volume, which retries over mounted shards.
+		return fmt.Errorf("failed to delete %d of %d original volume replicas for volume %d: %s",
+			len(deleteErrors), len(replicas), t.volumeID, strings.Join(deleteErrors, "; "))
 	}
 
+	t.GetLogger().WithFields(map[string]interface{}{
+		"volume_id":       t.volumeID,
+		"replica_count":   len(replicas),
+		"replica_servers": replicas,
+	}).Info("Successfully deleted volume from all replica servers")
+
 	return nil
 }
 

weed/worker/tasks/erasure_coding/ec_task_delete_swallow_test.go

@@ -0,0 +1,113 @@
+package erasure_coding
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// One reachable replica + one unreachable: the reachable delete still
+// succeeds, and the function surfaces an error naming the failure.
+func TestDeleteOriginalVolumeSurfacesReplicaFailures(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(91842)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 918420, 0x91842042)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("delete-surface-content-for-issue-9184"))
+	_ = framework.ReadAllAndClose(t, uploadResp)
+	require.Equal(t, http.StatusCreated, uploadResp.StatusCode)
+
+	task := NewErasureCodingTask(
+		"delete-surface-fix",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		"",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+
+	unreachable := "127.0.0.1:1"
+	task.sources = []*worker_pb.TaskSource{
+		{
+			Node:     clusterHarness.VolumeServerAddress(),
+			VolumeId: volumeID,
+		},
+		{
+			Node:     unreachable,
+			VolumeId: volumeID,
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	err := task.deleteOriginalVolume(ctx)
+	require.Error(t, err, "deleteOriginalVolume must surface replica delete failures (#9184)")
+	require.Contains(t, err.Error(), unreachable,
+		"returned error should name the replica that failed: %v", err)
+	require.True(t,
+		strings.Contains(err.Error(), "failed to delete"),
+		"returned error should describe what failed: %v", err)
+
+	_, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.Error(t, statusErr,
+		"reachable replica %d should have been deleted before failure was surfaced", volumeID)
+}
+
+func TestDeleteOriginalVolumeSucceedsWhenAllReplicasReachable(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(91844)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 918440, 0x91844042)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("delete-happy-path-content-for-issue-9184"))
+	_ = framework.ReadAllAndClose(t, uploadResp)
+	require.Equal(t, http.StatusCreated, uploadResp.StatusCode)
+
+	task := NewErasureCodingTask(
+		"delete-happy-path",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		"",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.sources = []*worker_pb.TaskSource{
+		{Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	require.NoError(t, task.deleteOriginalVolume(ctx))
+
+	_, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.Error(t, statusErr, "volume %d should be gone after successful delete", volumeID)
+}