EC worker: clear stale/interrupted shards at task start and on failure (#10738)

* EC worker: clear stale/interrupted shards at task start and on failure

The EC encode task cleared stale shards from a prior interrupted encode only
at 55% progress (after mark-readonly, copy, and generate), and used a
generation-fenced teardown. Two gaps left orphan shards behind:

  - a retried encode's prior attempt carries the same admin-issued encodeTsNs,
    and the server's teardown fence preserves same-or-newer generations, so the
    prior attempt's shards were never cleared;
  - shards left by an interrupted distribute often have an unreadable .vif
    generation (the sidecar never landed), which the fence also preserves.

Both survive the next volume-server restart as orphans and make detection
refuse the volume (Manual intervention required).

Move the cleanup to a Step 0 preflight that runs before any destructive step,
and switch it to the server's blanket (generation-independent) teardown -- the
same wipe the shell ec.encode pre-cleanup uses. The admin dedupe key already
prevents a concurrent newer encode of the volume, and the blanket path aborts
rather than clobber a live newer mount.

Add rollbackDistribute: a failure after distribute begins but before verify
commits the EC copy now tears down the shards it wrote and restores the sources
to writable, so a terminally-failed encode (a single-attempt job, or the last
of a retry series, which has no successor preflight) leaves nothing behind.

The preflight also rejects a plan with no targets or no source before marking
the source readonly.

* EC worker: reject malformed targets and keep source readonly on incomplete teardown

Address review feedback:

- ensureCleanEcStart only rejected an empty target slice; a target with an
  empty Node (or no shard ids) passed the length check, was then silently
  skipped by cleanupStaleEcShards, and let Execute mark the source readonly
  with nothing to distribute to. Validate each target before the first
  destructive step. Add regression cases.

- rollbackDistribute marked the source writable even when the shard teardown
  returned an error, exposing a writable source beside stale (possibly mounted)
  shards -- reads/writes could diverge and orphan cleanup will not remove a
  writable source. On an incomplete teardown, leave the source readonly for the
  next preflight or an operator to reconcile.

weed/worker/tasks/erasure_coding/ec_task.go

@@ -157,6 +157,17 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 		}
 	}()
 
+	// Step 0: Establish start-of-task invariants before any destructive step.
+	// Verify the plan is complete and clear EC shards left by a prior
+	// interrupted encode of this volume, so the encode begins from a clean
+	// slate. Failing here returns before the source is marked readonly or
+	// copied — nothing to roll back.
+	t.ReportProgressWithStage(5.0, "Verifying preconditions and clearing stale EC state")
+	t.GetLogger().Info("Verifying preconditions and clearing stale EC state")
+	if err := t.ensureCleanEcStart(ctx); err != nil {
+		return fmt.Errorf("EC preflight failed for volume %d: %w", t.volumeID, err)
+	}
+
 	// Step 1: Mark all replicas readonly, then reconcile them and select the most
 	// complete replica as the encode source. Encoding a stale replica and then
 	// deleting the originals would silently lose entries that exist only on another
@@ -197,15 +208,11 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 		return fmt.Errorf("failed to generate EC shards: %w", err)
 	}
 
-	// Clear partial EC shards left over on destinations from a prior failed
-	// encode so distributeEcShards' ReceiveFile is not refused by the
-	// mounted-volume guard.
-	t.ReportProgressWithStage(55.0, "Clearing stale EC shards on destinations")
-	t.GetLogger().Info("Clearing stale EC shards on destinations")
-	if err := t.cleanupStaleEcShards(ctx); err != nil {
-		t.rollbackReadonly(ctx)
-		return fmt.Errorf("failed to clear stale EC shards on destinations: %w", err)
-	}
+	// Stale EC shards from a prior interrupted encode were already cleared in
+	// the Step 0 preflight, before the source was marked readonly. The admin
+	// dedupe key (erasure_coding:<vid>:<collection>) prevents a concurrent
+	// same-volume encode, so no destination can regain stale shards between
+	// the preflight and distributeEcShards below.
 
 	// Delete 0-byte stub replicas left by an interrupted encode before the new
 	// EC files land. A stub shares the <collection>_<vid>.vif path the EC
@@ -219,10 +226,18 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 		return fmt.Errorf("failed to remove empty stub replicas: %w", err)
 	}
 
-	// Step 4: Distribute shards to destinations
+	// Step 4: Distribute shards to destinations.
+	// From here on a failure has written shards to destinations. Until verify
+	// passes we are not committed to the EC copy, so a failure must roll the
+	// attempt back — tear down the shards it distributed and restore the
+	// sources to writable — otherwise a terminally-failed encode (a
+	// single-attempt job, or the last of a retry series, which has no successor
+	// to clean up at its Step 0 preflight) strands orphan shards and a source
+	// fenced readonly.
 	t.ReportProgressWithStage(60.0, "Distributing EC shards to destinations")
 	t.GetLogger().Info("Distributing EC shards to destinations")
 	if err := t.distributeEcShards(shardFiles); err != nil {
+		t.rollbackDistribute(ctx)
 		return fmt.Errorf("failed to distribute EC shards: %w", err)
 	}
 
@@ -230,6 +245,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	t.ReportProgressWithStage(80.0, "Mounting EC shards")
 	t.GetLogger().Info("Mounting EC shards")
 	if err := t.mountEcShards(); err != nil {
+		t.rollbackDistribute(ctx)
 		return fmt.Errorf("failed to mount EC shards: %w", err)
 	}
 
@@ -238,8 +254,12 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	t.ReportProgressWithStage(85.0, "Verifying EC shards across destinations")
 	t.GetLogger().Info("Verifying EC shards across destinations")
 	if err := t.verifyEcShardsBeforeDelete(ctx); err != nil {
+		t.rollbackDistribute(ctx)
 		return fmt.Errorf("EC shard verification failed; refusing to delete source volume %d: %w", t.volumeID, err)
 	}
+	// Past verify the EC copy is recoverable; a Step 7 failure must NOT tear the
+	// shards down — the remaining source replicas are cleaned by the next
+	// detection's cleanupOrphanSourceReplicas instead.
 
 	// Step 7: Delete original volume
 	t.ReportProgressWithStage(90.0, "Deleting original volume")
@@ -917,16 +937,98 @@ func replicasPendingDelete(replicas []string, alreadyDeleted map[string]bool) []
 	return pending
 }
 
-// cleanupStaleEcShards unmounts and deletes any EC shards still mounted on
-// destinations from a previous failed encode of this volume. Targets every
-// node we plan to write to (t.targets) plus every node detection saw EC
-// shards on (t.sources with ShardIds set), and issues the cleanup over the
-// full shard range so a stale topology snapshot — or shards landed by a
-// prior attempt that haven't heartbeated yet — cannot leave the
-// mounted-volume guard tripped during distributeEcShards. Safe by ordering:
-// runs after the source .dat is in the worker's workdir and a full local
-// shard set is generated. Per-destination errors are aggregated, not
-// short-circuited.
+// ensureCleanEcStart runs first, before any destructive step, to establish
+// the invariants a fresh encode depends on:
+//   - a target set exists: an empty or malformed plan must fail here, not
+//     after the source has been marked readonly and copied;
+//   - a source replica exists to encode from;
+//   - no EC shards from a prior interrupted encode of this volume survive on
+//     the nodes this task will touch. Leftover partial shards trip the
+//     mounted-volume guard in distributeEcShards' ReceiveFile, are loaded as
+//     orphans on the next volume-server restart, and make detection refuse the
+//     volume ("Manual intervention required"). cleanupStaleEcShards blanket-
+//     wipes this volume's EC state on every touched node regardless of shard
+//     generation (a retried attempt's shards share this job's encodeTsNs, and
+//     an interrupted distribute often leaves shards with an unreadable
+//     generation — a fenced teardown would strand both).
+//
+// Cleaning at the start (rather than just before distribute) means the encode
+// begins from a clean slate and a preflight failure leaves the source
+// untouched — there is nothing to roll back. It is safe to delete stale shards
+// this early: the source's regular replica still holds the data until the
+// post-verify delete in Step 7.
+func (t *ErasureCodingTask) ensureCleanEcStart(ctx context.Context) error {
+	if len(t.targets) == 0 {
+		return fmt.Errorf("no EC shard targets for volume %d; refusing to mark source readonly", t.volumeID)
+	}
+	// A non-empty slice is not enough: a target with an empty Node (or no
+	// assigned shards) is silently skipped by cleanupStaleEcShards and by
+	// distributeEcShards, so a plan of only such entries would pass the length
+	// check and mark the source readonly before failing. Reject any malformed
+	// target here, before the first destructive step.
+	for i, target := range t.targets {
+		if target == nil || target.Node == "" || len(target.ShardIds) == 0 {
+			return fmt.Errorf("malformed EC shard target %d for volume %d; refusing to mark source readonly", i, t.volumeID)
+		}
+	}
+	if t.server == "" && len(t.getReplicas()) == 0 {
+		return fmt.Errorf("no source replica for volume %d", t.volumeID)
+	}
+	return t.cleanupStaleEcShards(ctx)
+}
+
+// rollbackDistribute undoes a failed attempt that had already begun writing EC
+// shards to destinations but had not yet committed to the EC copy (verify not
+// passed, so the sources are intact). It tears down the shards this attempt
+// distributed and restores the sources to writable, so a terminally-failed
+// encode — a single-attempt job, or the last of a retry series — leaves no
+// orphan shards and no source stuck readonly. On a retry the next attempt's
+// Step 0 preflight would also clear the shards, but the final attempt has no
+// successor; this makes every post-distribute failure self-cleaning.
+// cleanupStaleEcShards blanket-wipes this volume's EC state regardless of shard
+// generation — necessary because an interrupted distribute leaves shards whose
+// .vif generation is unreadable, which a fenced teardown would preserve.
+// Best-effort and uses a fresh context since the caller's may already be
+// cancelled (the very failure that brought us here).
+func (t *ErasureCodingTask) rollbackDistribute(_ context.Context) {
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	if err := t.cleanupStaleEcShards(ctx); err != nil {
+		// The teardown could not fully clear this volume's EC shards (e.g. an
+		// unreachable destination, or a shard that failed to unmount). Leave the
+		// source readonly rather than expose it for writes while stale shards
+		// linger: a writable source beside mounted stale shards would let reads
+		// and writes diverge, and orphan cleanup will not remove a writable
+		// source. The next encode's Step 0 preflight (or an operator) reconciles
+		// the state once the shards are reachable.
+		glog.Warningf("rollback: EC shard teardown incomplete for volume %d; leaving source readonly for reconciliation: %v", t.volumeID, err)
+		return
+	}
+	t.rollbackReadonly(ctx)
+}
+
+// cleanupStaleEcShards unmounts and deletes any EC shards for this volume on
+// destinations from a previous failed encode. Targets every node we plan to
+// write to (t.targets) plus every node detection saw EC shards on (t.sources
+// with ShardIds set), and issues the cleanup over the full shard range so a
+// stale topology snapshot — or shards landed by a prior attempt that haven't
+// heartbeated yet — cannot leave the mounted-volume guard tripped during
+// distributeEcShards. Called from the Step 0 preflight (ensureCleanEcStart) and
+// from rollbackDistribute.
+//
+// Teardown is UNFENCED (encodeTsNs=0 -> the server's blanket teardown), which
+// wipes every EC artifact for this volume on every disk regardless of
+// generation. A generation fence is wrong here for two reasons: (1) a retried
+// encode's prior attempt shares this job's encodeTsNs, and the server's fence
+// preserves same-or-newer, so a fenced teardown would strand it; (2) shards
+// left by an interrupted distribute often have an UNREADABLE .vif generation
+// (the sidecar never landed), which the fence also preserves. This is a
+// pre-encode / rollback wipe of a volume we are (re)encoding or abandoning, so
+// clearing all of its EC state is correct — the admin dedupe key
+// (erasure_coding:<vid>:<collection>) guarantees no concurrent newer encode of
+// this volume, and the blanket teardown's own replacement check aborts rather
+// than clobber a live newer mount. This mirrors the shell ec.encode pre-encode
+// cleanup. Per-destination errors are aggregated, not short-circuited.
 func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error {
 	nodes := make(map[string]struct{})
 	for _, source := range t.sources {
@@ -955,7 +1057,10 @@ func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error {
 			"shard_ids":   allShards,
 		}).Info("Clearing stale EC shards on destination before re-distribute")
 
-		if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards, t.encodeTsNs); err != nil {
+		// encodeTsNs=0 selects the server's blanket (generation-independent)
+		// teardown; see the function comment for why the fence is intentionally
+		// not used here.
+		if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards, 0); err != nil {
 			cleanupErrors = append(cleanupErrors, fmt.Sprintf("%s: %v", node, err))
 			t.GetLogger().WithFields(map[string]interface{}{
 				"volume_id":   t.volumeID,

weed/worker/tasks/erasure_coding/ec_task_preflight_test.go

@@ -0,0 +1,231 @@
+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"
+)
+
+// Reproduces the swtest interrupted-encode scenario at the task boundary: a
+// previous encode of this volume was cut mid-distribute, leaving partial EC
+// shards mounted on a destination. When a fresh encode task starts, its Step 0
+// preflight (ensureCleanEcStart) must clear those shards before any
+// destructive step, so distribute's ReceiveFile is not refused and no orphan
+// shards survive to confuse the volume-server loader or make detection refuse
+// the volume. The destination is named only as a target (no source row) to
+// prove the preflight reaches the full write set, not just shard-bearing
+// sources.
+func TestEnsureCleanEcStartClearsStaleShardsBeforeEncode(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(94782)
+		collection = "ec-preflight-invariant"
+	)
+
+	framework.AllocateVolume(t, grpcClient, volumeID, collection)
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 9478200, 0x9478BEEF)
+	upResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("payload-for-preflight-stale-ec-cleanup"))
+	_ = framework.ReadAllAndClose(t, upResp)
+	require.Equal(t, http.StatusCreated, upResp.StatusCode)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId: volumeID, Collection: collection,
+	})
+	require.NoError(t, err)
+
+	// A half-finished previous distribute left a partial shard set mounted.
+	staleShards := []uint32{0, 1, 2}
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId: volumeID, Collection: collection,
+		ShardIds: staleShards,
+	})
+	require.NoError(t, err)
+
+	shardPath := makeTinyEcShardFile(t)
+
+	// Precondition for the reproduction: the mounted partial EC blocks a fresh
+	// ReceiveFile via the mounted-volume guard.
+	err = sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath)
+	require.Error(t, err, "expected ReceiveFile to be refused while stale EC volume is mounted")
+
+	task := NewErasureCodingTask(
+		"preflight-invariant",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		collection,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	// Named only as a target — no source row for this node. The preflight must
+	// still clear its stale shards.
+	task.targets = []*worker_pb.TaskTarget{
+		{Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID, ShardIds: []uint32{0}},
+	}
+
+	require.NoError(t, task.ensureCleanEcStart(ctx),
+		"Step 0 preflight must clear stale EC shards at task start")
+
+	_, infoErr := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: volumeID})
+	require.Error(t, infoErr, "stale EC volume must be gone after the preflight")
+
+	require.NoError(t,
+		sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath),
+		"ReceiveFile must succeed once the preflight has cleared the stale shards")
+}
+
+// A post-distribute failure that has no successor to clean up after it (a
+// single-attempt job, or the last of a retry series) must leave nothing behind:
+// rollbackDistribute tears down the shards this attempt wrote while leaving the
+// source volume intact for a future re-encode. Reproduces the swtest finding
+// that a terminally-failed encode stranded orphan shards.
+func TestRollbackDistributeClearsShardsAndKeepsSource(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(94783)
+		collection = "ec-rollback-distribute"
+	)
+
+	framework.AllocateVolume(t, grpcClient, volumeID, collection)
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 9478300, 0x9478D00D)
+	upResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("payload-for-rollback-distribute"))
+	_ = framework.ReadAllAndClose(t, upResp)
+	require.Equal(t, http.StatusCreated, upResp.StatusCode)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId: volumeID, Collection: collection,
+	})
+	require.NoError(t, err)
+	// Shards this attempt "distributed" and mounted on the destination.
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId: volumeID, Collection: collection,
+		ShardIds: []uint32{0, 1, 2, 3},
+	})
+	require.NoError(t, err)
+
+	task := NewErasureCodingTask(
+		"rollback-distribute",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		collection,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.targets = []*worker_pb.TaskTarget{
+		{Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID, ShardIds: []uint32{0}},
+	}
+
+	// Simulate the failure path after distribute began.
+	task.rollbackDistribute(ctx)
+
+	_, infoErr := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: volumeID})
+	require.Error(t, infoErr, "rollback must tear down the shards this attempt distributed")
+
+	// The source normal volume must survive so a future re-encode can proceed.
+	_, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.NoError(t, statusErr, "rollback must leave the source volume intact")
+}
+
+// A malformed plan (no targets) must be rejected by the preflight before the
+// source is marked readonly or copied, so a bad plan cannot leave the source
+// fenced readonly with nothing to show for it. Pure precondition check: no
+// cluster and no RPC.
+func TestEnsureCleanEcStartRejectsMissingTargets(t *testing.T) {
+	task := NewErasureCodingTask(
+		"preflight-no-targets",
+		"10.0.0.1:8080",
+		42,
+		"c",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	// No targets set.
+	err := task.ensureCleanEcStart(context.Background())
+	require.Error(t, err, "preflight must reject an encode with no shard targets")
+	require.True(t, strings.Contains(err.Error(), "no EC shard targets"),
+		"error must name the missing-targets invariant, got: %v", err)
+}
+
+// A non-empty target slice whose entries are malformed (nil, empty Node, or no
+// assigned shards) must also be rejected before the source is marked readonly —
+// cleanupStaleEcShards silently skips such entries, so a length check alone
+// would let the encode proceed with nothing to distribute to.
+func TestEnsureCleanEcStartRejectsMalformedTarget(t *testing.T) {
+	cases := map[string][]*worker_pb.TaskTarget{
+		"nil target": {nil},
+		"empty node": {{Node: "", VolumeId: 42, ShardIds: []uint32{0}}},
+		"no shards":  {{Node: "10.0.0.2:8080", VolumeId: 42}},
+		"one good one bad": {
+			{Node: "10.0.0.2:8080", VolumeId: 42, ShardIds: []uint32{0}},
+			{Node: "", VolumeId: 42, ShardIds: []uint32{1}},
+		},
+	}
+	for name, targets := range cases {
+		t.Run(name, func(t *testing.T) {
+			task := NewErasureCodingTask(
+				"preflight-malformed-target",
+				"10.0.0.1:8080",
+				42,
+				"c",
+				grpc.WithTransportCredentials(insecure.NewCredentials()),
+			)
+			task.targets = targets
+			err := task.ensureCleanEcStart(context.Background())
+			require.Error(t, err, "preflight must reject a malformed target")
+			require.True(t, strings.Contains(err.Error(), "malformed EC shard target"),
+				"error must name the malformed-target invariant, got: %v", err)
+		})
+	}
+}
+
+// A plan with targets but no source replica and no assigned server must be
+// rejected before any destructive step.
+func TestEnsureCleanEcStartRejectsMissingSource(t *testing.T) {
+	task := NewErasureCodingTask(
+		"preflight-no-source",
+		"", // no assigned server
+		42,
+		"c",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.targets = []*worker_pb.TaskTarget{
+		{Node: "10.0.0.2:8080", VolumeId: 42, ShardIds: []uint32{0}},
+	}
+	// No sources and no server.
+	err := task.ensureCleanEcStart(context.Background())
+	require.Error(t, err, "preflight must reject an encode with no source replica")
+	require.True(t, strings.Contains(err.Error(), "no source replica"),
+		"error must name the missing-source invariant, got: %v", err)
+}