diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 2cbf9d669..54fbbf479 100644 --- a/weed/shell/command_ec_common.go +++ b/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 { diff --git a/weed/storage/erasure_coding/ec_teardown.go b/weed/storage/erasure_coding/ec_teardown.go new file mode 100644 index 000000000..b0e98af26 --- /dev/null +++ b/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 + }) +} diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index 27a44b444..ac8cbe787 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/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.