diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 74394dbdc..ff01027a8 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -455,6 +455,7 @@ message VolumeEcShardsDeleteRequest { bool full_teardown = 4; // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids } message VolumeEcShardsDeleteResponse { + bool full_teardown_done = 1; // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op } message VolumeEcShardsMountRequest { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 0ee561708..2cc21a31f 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -3669,9 +3669,10 @@ func (x *VolumeEcShardsDeleteRequest) GetFullTeardown() bool { } type VolumeEcShardsDeleteResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + FullTeardownDone bool `protobuf:"varint,1,opt,name=full_teardown_done,json=fullTeardownDone,proto3" json:"full_teardown_done,omitempty"` // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *VolumeEcShardsDeleteResponse) Reset() { @@ -3704,6 +3705,13 @@ func (*VolumeEcShardsDeleteResponse) Descriptor() ([]byte, []int) { return file_volume_server_proto_rawDescGZIP(), []int{65} } +func (x *VolumeEcShardsDeleteResponse) GetFullTeardownDone() bool { + if x != nil { + return x.FullTeardownDone + } + return false +} + type VolumeEcShardsMountRequest struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` @@ -7259,8 +7267,9 @@ const file_volume_server_proto_rawDesc = "" + "collection\x18\x02 \x01(\tR\n" + "collection\x12\x1b\n" + "\tshard_ids\x18\x03 \x03(\rR\bshardIds\x12#\n" + - "\rfull_teardown\x18\x04 \x01(\bR\ffullTeardown\"\x1e\n" + - "\x1cVolumeEcShardsDeleteResponse\"\xa0\x01\n" + + "\rfull_teardown\x18\x04 \x01(\bR\ffullTeardown\"L\n" + + "\x1cVolumeEcShardsDeleteResponse\x12,\n" + + "\x12full_teardown_done\x18\x01 \x01(\bR\x10fullTeardownDone\"\xa0\x01\n" + "\x1aVolumeEcShardsMountRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 126e41729..cd9cdbfa6 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -431,7 +431,7 @@ func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_se return nil, fmt.Errorf("full teardown of ec volume %d on %s: %w", req.VolumeId, location.Directory, err) } } - return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil + return &volume_server_pb.VolumeEcShardsDeleteResponse{FullTeardownDone: true}, nil } glog.V(0).Infof("ec volume %s shard delete %v", bName, req.ShardIds) diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 7afc21cd6..c0c709365 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "fmt" "regexp" "slices" @@ -536,6 +537,13 @@ func sourceServerDeleteEcShards(grpcDialOption grpc.DialOption, collection strin } +// errFullTeardownNotAcked marks a VolumeEcShardsDelete whose server did not +// echo full_teardown_done -- a pre-upgrade volume server silently ignoring the +// flag. The pre-encode sweep treats it as fatal only for a reported (mounted) +// leftover; for an unreported node it is best-effort, since that node likely had +// nothing to wipe and an old server cannot acknowledge either way. +var errFullTeardownNotAcked = errors.New("volume server did not perform full teardown") + // unmountAndDeleteEcShardsQuiet unmounts then deletes shards on one server in a // single connection, without the per-call logging the interactive helpers emit. // Used by the orphan sweep, which fans out to every node x volume and would @@ -549,14 +557,18 @@ func unmountAndDeleteEcShardsQuiet(grpcDialOption grpc.DialOption, collection st }); err != nil { return fmt.Errorf("unmount: %w", err) } - if _, err := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{ + resp, err := volumeServerClient.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{ VolumeId: uint32(volumeId), Collection: collection, ShardIds: ids, FullTeardown: true, - }); err != nil { + }) + if err != nil { return fmt.Errorf("delete: %w", err) } + if !resp.GetFullTeardownDone() { + return fmt.Errorf("delete on %s: %w", location, errFullTeardownNotAcked) + } return nil }) } diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index a91aab188..f5918adee 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" @@ -405,12 +406,14 @@ func clearPreexistingEcShards(commandEnv *CommandEnv, topologyInfo *master_pb.To ewg.Add(func() error { if err := unmountAndDeleteEcShardsQuiet(commandEnv.option.GrpcDialOption, collection, vid, addr, allShardIds); err != nil { // Surface a reachable node whose delete genuinely failed (its orphan would - // be re-stamped by a later copy installing the new .vif); stay best-effort - // only for an unreachable node, which cannot receive this new generation. - if fatal || !isNodeUnreachable(err) { + // be re-stamped by a later copy installing the new .vif). Stay best-effort + // for an unreachable node (it cannot receive this new generation) and for + // a pre-upgrade node that did not ack full_teardown on an UNREPORTED pair + // (it likely had nothing to wipe); a reported leftover stays fatal. + if fatal || (!isNodeUnreachable(err) && !errors.Is(err, errFullTeardownNotAcked)) { return fmt.Errorf("clear stale ec shards for volume %d on %s: %w", vid, addr, err) } - glog.V(1).Infof("orphan sweep: volume %d on %s unreachable, skipping: %v", vid, addr, err) + glog.V(1).Infof("orphan sweep: volume %d on %s skipped: %v", vid, addr, err) } return nil }) diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index fcc5921f5..7bdf201f5 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -977,14 +977,18 @@ func unmountAndDeleteEcShards( }); err != nil { return fmt.Errorf("unmount: %w", err) } - if _, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{ + resp, err := client.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{ VolumeId: volumeID, Collection: collection, ShardIds: shardIds, FullTeardown: true, - }); err != nil { + }) + 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 }) }