erasure_coding: share the EC shard teardown primitive (#10740)

The unmount+full-teardown of EC shards was duplicated: the plugin-worker EC
task had unmountAndDeleteEcShards and the shell had unmountAndDeleteEcShardsQuiet,
byte-identical apart from a fence parameter and a sentinel error. That
duplication is how the teardown fence semantics drifted between the two paths.

Distribute, mount and verify already live in weed/storage/erasure_coding and are
shared by both callers; move the teardown there too, as UnmountAndDeleteEcShards
plus the shared ErrFullTeardownNotAcked sentinel. Both paths now call the one
function, so the fence semantics cannot diverge again. The shell keeps a thin
type-converting wrapper and aliases the sentinel; behavior is unchanged.

weed/shell/command_ec_common.go

@@ -2,7 +2,6 @@ package shell
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"os"
 	"regexp"
@@ -561,8 +560,9 @@ func sourceServerDeleteEcShards(grpcDialOption grpc.DialOption, collection strin
 // errFullTeardownNotAcked marks a reachable server that completed the delete RPC
 // but did not report full_teardown_done (a pre-upgrade volume server). The orphan
 // sweep must treat this as fatal: the node may still hold an orphan that a later
-// copy would re-stamp into the new generation.
-var errFullTeardownNotAcked = errors.New("delete did not perform full teardown (pre-upgrade volume server?); a stale EC generation may remain")
+// copy would re-stamp into the new generation. Aliased to the shared sentinel so
+// the shell and the plugin-worker EC task agree on the teardown-not-acked signal.
+var errFullTeardownNotAcked = erasure_coding.ErrFullTeardownNotAcked
 
 // pingVolumeServer probes node liveness with an empty-target Ping, which is never
 // maintenance-gated, and returns the raw Ping error (nil on success). It lets the
@@ -586,28 +586,8 @@ func pingVolumeServer(grpcDialOption grpc.DialOption, location pb.ServerAddress)
 // Used by the orphan sweep, which fans out to every node x volume and would
 // otherwise flood the shell with no-op lines.
 func unmountAndDeleteEcShardsQuiet(grpcDialOption grpc.DialOption, collection string, volumeId needle.VolumeId, location pb.ServerAddress, shardIds []erasure_coding.ShardId) error {
-	ids := erasure_coding.ShardIdsToUint32(shardIds)
-	return operation.WithVolumeServerClient(false, location, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-		if _, err := volumeServerClient.VolumeEcShardsUnmount(context.Background(), &volume_server_pb.VolumeEcShardsUnmountRequest{
-			VolumeId: uint32(volumeId),
-			ShardIds: ids,
-		}); err != nil {
-			return fmt.Errorf("unmount: %w", err)
-		}
-		resp, err := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{
-			VolumeId:     uint32(volumeId),
-			Collection:   collection,
-			ShardIds:     ids,
-			FullTeardown: true,
-		})
-		if err != nil {
-			return fmt.Errorf("delete: %w", err)
-		}
-		if !resp.GetFullTeardownDone() {
-			return fmt.Errorf("delete on %s: %w", location, errFullTeardownNotAcked)
-		}
-		return nil
-	})
+	return erasure_coding.UnmountAndDeleteEcShards(context.Background(), grpcDialOption, location, collection,
+		uint32(volumeId), erasure_coding.ShardIdsToUint32(shardIds), 0)
 }
 
 func unmountEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceLocation pb.ServerAddress, toBeUnmountedShardIds []erasure_coding.ShardId) error {

weed/storage/erasure_coding/ec_teardown.go

@@ -0,0 +1,76 @@
+package erasure_coding
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"google.golang.org/grpc"
+)
+
+// ErrFullTeardownNotAcked marks a reachable server that completed the delete
+// RPC but did not report a full teardown (e.g. a pre-upgrade volume server), so
+// a stale EC generation may remain on it. Callers distinguish this from an
+// unreachable node (which may recover and be re-swept) with errors.Is.
+var ErrFullTeardownNotAcked = errors.New("delete did not perform full teardown (pre-upgrade volume server?); a stale EC generation may remain")
+
+// UnmountAndDeleteEcShards unmounts then tears down the named EC shards for a
+// volume on one server. Unmount must precede delete (delete requires the shard
+// be unmounted); both RPCs are idempotent against missing shards.
+//
+// encodeTsNs fences both RPCs:
+//   - 0 selects the server's blanket, generation-independent teardown. This is
+//     the correct choice for a pre-encode or rollback wipe: it clears same-
+//     generation shards (a retried encode's prior attempt shares the job's
+//     generation) and shards whose .vif generation is unreadable (an
+//     interrupted distribute never landed the sidecar) — both of which a fenced
+//     teardown preserves. The blanket path aborts rather than clobber a live
+//     newer mount, and the caller must guarantee no concurrent newer encode of
+//     this volume (e.g. the admin dedupe key, or an operator lock).
+//   - a non-zero value fences the teardown to strictly-older generations,
+//     preserving same-or-newer, generation 0, and an unreadable .vif — for a
+//     stale-worker cleanup that must never wipe a newer run's live shards.
+//
+// Returns ErrFullTeardownNotAcked (wrapped, so errors.Is matches) when a
+// reachable server does not ack the full teardown.
+//
+// This is the single teardown primitive shared by the plugin-worker EC task
+// and the shell ec.encode pre-cleanup, so the fence semantics cannot drift
+// between the two paths.
+func UnmountAndDeleteEcShards(
+	ctx context.Context,
+	dialOption grpc.DialOption,
+	server pb.ServerAddress,
+	collection string,
+	volumeID uint32,
+	shardIds []uint32,
+	encodeTsNs int64,
+) error {
+	return operation.WithVolumeServerClient(false, server, dialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			if _, err := client.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{
+				VolumeId:   volumeID,
+				ShardIds:   shardIds,
+				EncodeTsNs: encodeTsNs,
+			}); err != nil {
+				return fmt.Errorf("unmount: %w", err)
+			}
+			resp, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
+				VolumeId:     volumeID,
+				Collection:   collection,
+				ShardIds:     shardIds,
+				FullTeardown: true,
+				EncodeTsNs:   encodeTsNs,
+			})
+			if err != nil {
+				return fmt.Errorf("delete: %w", err)
+			}
+			if !resp.GetFullTeardownDone() {
+				return fmt.Errorf("delete on %s: %w", server, ErrFullTeardownNotAcked)
+			}
+			return nil
+		})
+}

weed/worker/tasks/erasure_coding/ec_task.go

@@ -1058,9 +1058,9 @@ func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error {
 		}).Info("Clearing stale EC shards on destination before re-distribute")
 
 		// 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 {
+		// teardown; see erasure_coding.UnmountAndDeleteEcShards for why the
+		// fence is intentionally not used here.
+		if err := erasure_coding.UnmountAndDeleteEcShards(ctx, t.grpcDialOption, pb.ServerAddress(node), t.collection, t.volumeID, allShards, 0); err != nil {
 			cleanupErrors = append(cleanupErrors, fmt.Sprintf("%s: %v", node, err))
 			t.GetLogger().WithFields(map[string]interface{}{
 				"volume_id":   t.volumeID,
@@ -1095,47 +1095,6 @@ func fullShardIdRange(dataShards, parityShards int32) []uint32 {
 	return ids
 }
 
-// unmountAndDeleteEcShards unmounts then deletes the named shards on one
-// destination. Unmount must precede delete (delete requires the shard be
-// unmounted); both RPCs are idempotent against missing shards.
-func unmountAndDeleteEcShards(
-	ctx context.Context,
-	dialOption grpc.DialOption,
-	destination string,
-	volumeID uint32,
-	collection string,
-	shardIds []uint32,
-	encodeTsNs int64,
-) error {
-	return operation.WithVolumeServerClient(false, pb.ServerAddress(destination), dialOption,
-		func(client volume_server_pb.VolumeServerClient) error {
-			// encodeTsNs fences both RPCs against a newer run on a shared node: the
-			// server skips a disk whose mounted/on-disk generation is same-or-newer.
-			// 0 (legacy/shell) leaves the unconditional unmount + blanket teardown.
-			if _, err := client.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{
-				VolumeId:   volumeID,
-				ShardIds:   shardIds,
-				EncodeTsNs: encodeTsNs,
-			}); err != nil {
-				return fmt.Errorf("unmount: %w", err)
-			}
-			resp, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
-				VolumeId:     volumeID,
-				Collection:   collection,
-				ShardIds:     shardIds,
-				FullTeardown: true,
-				EncodeTsNs:   encodeTsNs,
-			})
-			if err != nil {
-				return fmt.Errorf("delete: %w", err)
-			}
-			if !resp.GetFullTeardownDone() {
-				return fmt.Errorf("delete: %s did not perform full teardown (pre-upgrade volume server?); a stale EC generation may remain", destination)
-			}
-			return nil
-		})
-}
-
 // verifyDatIdxConsistency checks that all .idx entries reference data within the
 // .dat file. Since .dat and .idx are copied as separate network transfers, the
 // .idx may have entries from writes that landed after the .dat was copied.