From 05b4b5bf569a86d0d416c124e216098d17b4328b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 30 Jun 2026 23:20:50 -0700 Subject: [PATCH] ec: expose force_deleted_needles_check in ScrubEcVolume RPC and shell (#10176) * ec: expose force_deleted_needles_check in ScrubEcVolume RPC and shell FULL EC scrubs can opt into strict deleted-needle verification via the -forceDeletedNeedlesCheck shell flag, off by default since it can report false positives when EC indexes disagree. Rejected for non-FULL modes. The Rust volume server parses the new field and ignores it: its FULL scrub verifies shards via RS parity, not per-needle reads. * volume: require admin auth for ScrubEcVolume ScrubEcVolume ran unauthenticated while its sibling ScrubVolume, and the rest of the mutating volume handlers, gate on checkGrpcAdminAuth. Close the gap so an EC scrub can't be triggered anonymously. * shell: reject ec.scrub -forceDeletedNeedlesCheck outside full mode Fail in the client before fanning out to every volume server, instead of erroring halfway through once the servers reject the request. --- seaweed-volume/proto/volume_server.proto | 1 + weed/pb/volume_server.proto | 1 + weed/pb/volume_server_pb/volume_server.pb.go | 19 +++++++++++++----- weed/server/volume_grpc_scrub.go | 10 ++++++++-- weed/shell/command_ec_scrub.go | 21 +++++++++++++------- 5 files changed, 38 insertions(+), 14 deletions(-) diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index da1b941bd..deb34228d 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -708,6 +708,7 @@ message ScrubEcVolumeRequest { VolumeScrubMode mode = 1; // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. repeated uint32 volume_ids = 2; + bool force_deleted_needles_check = 3; // FULL mode only; may report false positives when EC indexes disagree } message ScrubEcVolumeResponse { uint64 total_volumes = 1; diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 889b4dc73..abf9da1cf 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -711,6 +711,7 @@ message ScrubEcVolumeRequest { VolumeScrubMode mode = 1; // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. repeated uint32 volume_ids = 2; + bool force_deleted_needles_check = 3; // FULL mode only; may report false positives when EC indexes disagree } message ScrubEcVolumeResponse { uint64 total_volumes = 1; diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index d02b35bb1..0bc292e23 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -6018,9 +6018,10 @@ type ScrubEcVolumeRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Mode VolumeScrubMode `protobuf:"varint,1,opt,name=mode,proto3,enum=volume_server_pb.VolumeScrubMode" json:"mode,omitempty"` // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. - VolumeIds []uint32 `protobuf:"varint,2,rep,packed,name=volume_ids,json=volumeIds,proto3" json:"volume_ids,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + VolumeIds []uint32 `protobuf:"varint,2,rep,packed,name=volume_ids,json=volumeIds,proto3" json:"volume_ids,omitempty"` + ForceDeletedNeedlesCheck bool `protobuf:"varint,3,opt,name=force_deleted_needles_check,json=forceDeletedNeedlesCheck,proto3" json:"force_deleted_needles_check,omitempty"` // FULL mode only; may report false positives when EC indexes disagree + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ScrubEcVolumeRequest) Reset() { @@ -6067,6 +6068,13 @@ func (x *ScrubEcVolumeRequest) GetVolumeIds() []uint32 { return nil } +func (x *ScrubEcVolumeRequest) GetForceDeletedNeedlesCheck() bool { + if x != nil { + return x.ForceDeletedNeedlesCheck + } + return false +} + type ScrubEcVolumeResponse struct { state protoimpl.MessageState `protogen:"open.v1"` TotalVolumes uint64 `protobuf:"varint,1,opt,name=total_volumes,json=totalVolumes,proto3" json:"total_volumes,omitempty"` @@ -7510,11 +7518,12 @@ const file_volume_server_proto_rawDesc = "" + "\vtotal_files\x18\x02 \x01(\x04R\n" + "totalFiles\x12*\n" + "\x11broken_volume_ids\x18\x03 \x03(\rR\x0fbrokenVolumeIds\x12\x18\n" + - "\adetails\x18\x04 \x03(\tR\adetails\"l\n" + + "\adetails\x18\x04 \x03(\tR\adetails\"\xab\x01\n" + "\x14ScrubEcVolumeRequest\x125\n" + "\x04mode\x18\x01 \x01(\x0e2!.volume_server_pb.VolumeScrubModeR\x04mode\x12\x1d\n" + "\n" + - "volume_ids\x18\x02 \x03(\rR\tvolumeIds\"\xf0\x01\n" + + "volume_ids\x18\x02 \x03(\rR\tvolumeIds\x12=\n" + + "\x1bforce_deleted_needles_check\x18\x03 \x01(\bR\x18forceDeletedNeedlesCheck\"\xf0\x01\n" + "\x15ScrubEcVolumeResponse\x12#\n" + "\rtotal_volumes\x18\x01 \x01(\x04R\ftotalVolumes\x12\x1f\n" + "\vtotal_files\x18\x02 \x01(\x04R\n" + diff --git a/weed/server/volume_grpc_scrub.go b/weed/server/volume_grpc_scrub.go index 3b967e35c..412f0bbd2 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -93,6 +93,13 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S } func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb.ScrubEcVolumeRequest) (*volume_server_pb.ScrubEcVolumeResponse, error) { + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return nil, err + } + if req.GetForceDeletedNeedlesCheck() && req.GetMode() != volume_server_pb.VolumeScrubMode_FULL { + return nil, fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + } + vids := []needle.VolumeId{} if len(req.GetVolumeIds()) == 0 { for _, l := range vs.store.Locations { @@ -124,8 +131,7 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb case volume_server_pb.VolumeScrubMode_LOCAL: files, shardInfos, serrs = v.ScrubLocal() case volume_server_pb.VolumeScrubMode_FULL: - // TODO: expose force_deleted_needles_check in the RPC and weed shell. - files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, false) + files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, req.GetForceDeletedNeedlesCheck()) case volume_server_pb.VolumeScrubMode_CHECKSUM: // Verify each local shard's raw bytes against the bitrot sidecar, // exercising cold parity shards. Read-only. ChecksumScrub's first diff --git a/weed/shell/command_ec_scrub.go b/weed/shell/command_ec_scrub.go index e9bd36f6e..13b6897ab 100644 --- a/weed/shell/command_ec_scrub.go +++ b/weed/shell/command_ec_scrub.go @@ -20,11 +20,12 @@ func init() { } type commandEcVolumeScrub struct { - env *CommandEnv - volumeServerAddrs []pb.ServerAddress - volumeIDs []uint32 - mode volume_server_pb.VolumeScrubMode - grpcDialOption grpc.DialOption + env *CommandEnv + volumeServerAddrs []pb.ServerAddress + volumeIDs []uint32 + mode volume_server_pb.VolumeScrubMode + forceDeletedNeedlesCheck bool + grpcDialOption grpc.DialOption } func (c *commandEcVolumeScrub) Name() string { @@ -52,6 +53,7 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer mode := volScrubCommand.String("mode", "local", "scrubbing mode (index/local/full/checksum)") maxParallelization := volScrubCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") showDetails := volScrubCommand.Bool("details", false, "display scrub result details, if available") + forceDeletedNeedlesCheck := volScrubCommand.Bool("forceDeletedNeedlesCheck", false, "force strict verification of deleted needles (full mode only); may report false positives when EC indexes disagree") if err = volScrubCommand.Parse(args); err != nil { return err @@ -104,6 +106,10 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer } fmt.Fprintf(writer, "using %s mode\n", c.mode.String()) c.env = commandEnv + c.forceDeletedNeedlesCheck = *forceDeletedNeedlesCheck + if c.forceDeletedNeedlesCheck && c.mode != volume_server_pb.VolumeScrubMode_FULL { + return fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + } return c.scrubEcVolumes(writer, *maxParallelization, *showDetails) } @@ -125,8 +131,9 @@ func (c *commandEcVolumeScrub) scrubEcVolumes(writer io.Writer, maxParallelizati err := operation.WithVolumeServerClient(false, addr, c.env.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { res, err := volumeServerClient.ScrubEcVolume(context.Background(), &volume_server_pb.ScrubEcVolumeRequest{ - Mode: c.mode, - VolumeIds: c.volumeIDs, + Mode: c.mode, + VolumeIds: c.volumeIDs, + ForceDeletedNeedlesCheck: c.forceDeletedNeedlesCheck, }) if err != nil { return err