diff --git a/weed/pb/worker.proto b/weed/pb/worker.proto index 00d4cb288..1505b128c 100644 --- a/weed/pb/worker.proto +++ b/weed/pb/worker.proto @@ -395,6 +395,10 @@ message EcBalanceTaskParams { int32 max_parallelization = 2; // Max parallel shard moves within a batch int32 timeout_seconds = 3; // Operation timeout per move repeated EcShardMoveSpec moves = 4; // Batch: multiple shard moves in one job + // For a dedup move, the node that keeps the shard. The worker confirms this + // node really holds it before deleting the copy, so a topology entry naming a + // location that holds nothing cannot cause the last copy to be removed. + string dedup_keep_node = 5; } // EcShardMoveSpec describes a single EC shard move within a batch diff --git a/weed/pb/worker_pb/worker.pb.go b/weed/pb/worker_pb/worker.pb.go index d3305a474..a4048c51d 100644 --- a/weed/pb/worker_pb/worker.pb.go +++ b/weed/pb/worker_pb/worker.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v6.33.4 +// protoc v7.35.0 // source: worker.proto package worker_pb @@ -3150,8 +3150,12 @@ type EcBalanceTaskParams struct { MaxParallelization int32 `protobuf:"varint,2,opt,name=max_parallelization,json=maxParallelization,proto3" json:"max_parallelization,omitempty"` // Max parallel shard moves within a batch TimeoutSeconds int32 `protobuf:"varint,3,opt,name=timeout_seconds,json=timeoutSeconds,proto3" json:"timeout_seconds,omitempty"` // Operation timeout per move Moves []*EcShardMoveSpec `protobuf:"bytes,4,rep,name=moves,proto3" json:"moves,omitempty"` // Batch: multiple shard moves in one job - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // For a dedup move, the node that keeps the shard. The worker confirms this + // node really holds it before deleting the copy, so a topology entry naming a + // location that holds nothing cannot cause the last copy to be removed. + DedupKeepNode string `protobuf:"bytes,5,opt,name=dedup_keep_node,json=dedupKeepNode,proto3" json:"dedup_keep_node,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *EcBalanceTaskParams) Reset() { @@ -3212,6 +3216,13 @@ func (x *EcBalanceTaskParams) GetMoves() []*EcShardMoveSpec { return nil } +func (x *EcBalanceTaskParams) GetDedupKeepNode() string { + if x != nil { + return x.DedupKeepNode + } + return "" +} + // EcShardMoveSpec describes a single EC shard move within a batch type EcShardMoveSpec struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4248,12 +4259,13 @@ const file_worker_proto_rawDesc = "" + "\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" + "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\"I\n" + "\x15ReplicationTaskConfig\x120\n" + - "\x14target_replica_count\x18\x01 \x01(\x05R\x12targetReplicaCount\"\xbe\x01\n" + + "\x14target_replica_count\x18\x01 \x01(\x05R\x12targetReplicaCount\"\xe6\x01\n" + "\x13EcBalanceTaskParams\x12\x1b\n" + "\tdisk_type\x18\x01 \x01(\tR\bdiskType\x12/\n" + "\x13max_parallelization\x18\x02 \x01(\x05R\x12maxParallelization\x12'\n" + "\x0ftimeout_seconds\x18\x03 \x01(\x05R\x0etimeoutSeconds\x120\n" + - "\x05moves\x18\x04 \x03(\v2\x1a.worker_pb.EcShardMoveSpecR\x05moves\"\xf7\x01\n" + + "\x05moves\x18\x04 \x03(\v2\x1a.worker_pb.EcShardMoveSpecR\x05moves\x12&\n" + + "\x0fdedup_keep_node\x18\x05 \x01(\tR\rdedupKeepNode\"\xf7\x01\n" + "\x0fEcShardMoveSpec\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" + "\bshard_id\x18\x02 \x01(\rR\ashardId\x12\x1e\n" + diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index c222b6116..96d03cb81 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -1089,6 +1089,28 @@ func (ecb *ecBalancer) executePhase(byID map[string]*EcNode, moves []ecbalancer. return ewg.Wait() } +// verifyEcShardOnKeepNode confirms the node a dedup move chose to keep actually +// holds the shard, so a duplicate is only removed when a real copy remains. An +// unreachable keep node is unknown, not confirmed, and blocks the delete — as +// does one that answers too slowly to be waited on, which is why this is +// bounded rather than left to hang the whole balance run. +// ecShardVerifyTimeout bounds the keep-node inventory query. A node that accepts +// the connection but never answers must not stall the whole balance run. +const ecShardVerifyTimeout = 30 * time.Second + +func verifyEcShardOnKeepNode(grpcDialOption grpc.DialOption, collection string, vid needle.VolumeId, keepNode string, shardId erasure_coding.ShardId) error { + if keepNode == "" { + return fmt.Errorf("refusing dedup delete of %d.%d: no keep node recorded", vid, shardId) + } + ctx, cancel := context.WithTimeout(context.Background(), ecShardVerifyTimeout) + defer cancel() + if err := erasure_coding.VerifyShardsOnServer(ctx, collection, uint32(vid), keepNode, + []uint32{uint32(shardId)}, grpcDialOption); err != nil { + return fmt.Errorf("refusing dedup delete: %w", err) + } + return nil +} + func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) error { src := byID[m.SourceNode] if src == nil { @@ -1105,6 +1127,13 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e return nil } grpcDialOption := ecb.commandEnv.option.GrpcDialOption + // Nothing is copied first, so the shard surviving elsewhere is the only + // thing making this safe -- and the plan saying so is not evidence. A + // topology entry can name a location holding nothing, and deleting on + // that basis removes the last copy. Confirm the keep node has it. + if err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId); err != nil { + return err + } addr := pb.NewServerAddressFromDataNode(src.info) if err := unmountEcShards(grpcDialOption, vid, addr, shardIds); err != nil { return err diff --git a/weed/storage/erasure_coding/ecbalancer/balancer.go b/weed/storage/erasure_coding/ecbalancer/balancer.go index 83fcd8f6a..74cf233b2 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer.go @@ -74,6 +74,10 @@ type Move struct { TargetDisk uint32 TargetRack string Phase string // "dedup", "cross_rack", "within_rack", "global" + // KeepNode is set for dedup moves: the node the plan chose to keep the + // shard on. The executor confirms it really holds the shard before + // deleting the copy. + KeepNode string } // Options tunes a Plan run. @@ -114,6 +118,7 @@ type move struct { target *Node targetDisk uint32 phase string + keepNode string } // NewTopology returns an empty topology to populate. @@ -270,6 +275,7 @@ func Plan(topo *Topology, opts Options) []Move { TargetDisk: m.targetDisk, TargetRack: m.target.rack, Phase: m.phase, + KeepNode: m.keepNode, }) } return out @@ -318,6 +324,7 @@ func detectDuplicateShards(vk volKey, nodes map[string]*Node) []*move { } return locs[i].id < locs[j].id }) + keep := locs[len(locs)-1] for _, node := range locs[:len(locs)-1] { moves = append(moves, &move{ volumeID: vk.vid, @@ -328,6 +335,7 @@ func detectDuplicateShards(vk volKey, nodes map[string]*Node) []*move { target: node, targetDisk: shardDiskID(node, vk, shardID), phase: "dedup", + keepNode: keep.id, }) } } diff --git a/weed/storage/erasure_coding/verification.go b/weed/storage/erasure_coding/verification.go index caf540154..390e33a93 100644 --- a/weed/storage/erasure_coding/verification.go +++ b/weed/storage/erasure_coding/verification.go @@ -137,3 +137,51 @@ func SummarizeShardInventory(perServer map[string]ServerShardInventory) string { } return string(b) } + +// VerifyShardsOnServer confirms one server really holds the named shards of a +// specific (collection, volume), for callers about to delete another copy. +// +// Collection is checked, unlike in VerifyShardsAcrossServers: the inventory RPC +// is keyed by volume id alone, so a server answering for volume N says nothing +// about which collection's volume N it means. Approving a delete on the +// strength of a different collection's shard would remove the last real copy — +// the exact outcome the caller is trying to prevent. +// +// A server that cannot be queried is unknown, not confirmed, and returns an +// error: treating an unreachable peer as proof of a surviving copy is how a +// network blip becomes data loss. +func VerifyShardsOnServer(ctx context.Context, collection string, volumeID uint32, + server string, shardIDs []uint32, dialOption grpc.DialOption) error { + + if server == "" { + return fmt.Errorf("no server given to verify volume %d shard(s) %v", volumeID, shardIDs) + } + + var present ShardBits + callErr := operation.WithVolumeServerClient(false, pb.ServerAddress(server), dialOption, + func(client volume_server_pb.VolumeServerClient) error { + resp, e := client.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{ + VolumeId: volumeID, + }) + if e != nil { + return e + } + for _, s := range resp.EcShardInfos { + if s.VolumeId != volumeID || s.Collection != collection || s.ShardId >= MaxShardCount { + continue + } + present = present.Set(ShardId(s.ShardId)) + } + return nil + }) + if callErr != nil { + return fmt.Errorf("verify volume %d shard(s) %v on %s: %w", volumeID, shardIDs, server, callErr) + } + + for _, sid := range shardIDs { + if !present.Has(ShardId(sid)) { + return fmt.Errorf("%s does not hold ec shard %d.%d of collection %q", server, volumeID, sid, collection) + } + } + return nil +} diff --git a/weed/worker/tasks/ec_balance/dedup_safety_test.go b/weed/worker/tasks/ec_balance/dedup_safety_test.go new file mode 100644 index 000000000..350c238b3 --- /dev/null +++ b/weed/worker/tasks/ec_balance/dedup_safety_test.go @@ -0,0 +1,128 @@ +package ec_balance + +import ( + "context" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" +) + +// The dedup phase deletes a shard it believes is duplicated elsewhere. It picks +// its victims from the master's topology, and the topology can name a location +// that holds nothing: a volume server answering "not found ec volume id N" for +// a shard the master lists is the observable form of that. Deleting on the +// strength of a phantom peer removes the only real copy. +// +// The move phase already refuses to work on trust — it verifies the shard +// registered on the destination before removing the source. These tests hold +// dedup to the same standard, because it is the more dangerous of the two: it +// deletes without copying anything first. + +const ( + dedupTestVolumeID = uint32(42561) + dedupTestCollection = "pm-itatiaiucu-01" +) + +func dedupParams(sourceNode, keepNode string, shardID uint32) *worker_pb.TaskParams { + // Dedup is signalled by source and target being the same node and disk. + loc := &worker_pb.TaskSource{Node: sourceNode, DiskId: 0, ShardIds: []uint32{shardID}} + return &worker_pb.TaskParams{ + VolumeId: dedupTestVolumeID, + Collection: dedupTestCollection, + Sources: []*worker_pb.TaskSource{loc}, + Targets: []*worker_pb.TaskTarget{{Node: sourceNode, DiskId: 0, ShardIds: []uint32{shardID}}}, + // The dedup branch is only reachable when EC params are present; without + // them the job silently takes the copy-and-move path instead. + TaskParams: &worker_pb.TaskParams_EcBalanceParams{ + EcBalanceParams: &worker_pb.EcBalanceTaskParams{DedupKeepNode: keepNode}, + }, + } +} + +func newDedupTask() *ECBalanceTask { + return NewECBalanceTask("dedup-test", dedupTestVolumeID, dedupTestCollection, + grpc.WithTransportCredentials(insecure.NewCredentials())) +} + +// The failure that cost this cluster shards: the topology claims shard 0 lives +// on two nodes, only one of them actually has it, and dedup is pointed at the +// one that does. Nothing else in the cluster holds shard 0, so deleting it is +// unrecoverable — and no copy was made first, because dedup never makes one. +func TestDedupRefusesToDeleteTheOnlyRealCopy(t *testing.T) { + holder := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + // The peer the topology believes also has shard 0. Its disks are empty, + // exactly like a server answering "not found ec volume id". + phantom := startFakeEcVolumeServer(t, dedupTestVolumeID) + + if !holder.has(dedupTestVolumeID, 0) { + t.Fatal("precondition: holder should start with shard 0") + } + if phantom.has(dedupTestVolumeID, 0) { + t.Fatal("precondition: phantom should hold nothing") + } + + err := newDedupTask().Execute(context.Background(), dedupParams(holder.address(), phantom.address(), 0)) + + if holder.has(dedupTestVolumeID, 0) { + return // refused, or verified and backed off — either is safe + } + t.Fatalf("dedup deleted the only copy of shard %d.0 (deleted=%v, err=%v); "+ + "the surviving copy was never confirmed to exist", + dedupTestVolumeID, holder.deletedShards(), err) +} + +// A shard genuinely present on two servers is safe to thin out: one copy is +// redundant and removing it loses nothing. Dedup has to keep working here, or +// the guard above has simply disabled the feature. +func TestDedupRemovesAGenuineDuplicate(t *testing.T) { + victim := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + survivor := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + + if err := newDedupTask().Execute(context.Background(), dedupParams(victim.address(), survivor.address(), 0)); err != nil { + t.Fatalf("dedup of a real duplicate should succeed, got %v", err) + } + if victim.has(dedupTestVolumeID, 0) { + t.Error("dedup left the duplicate in place; capacity is not reclaimed") + } + if !survivor.has(dedupTestVolumeID, 0) { + t.Error("dedup removed the surviving copy instead of the duplicate") + } +} + +// An unreachable peer is not evidence of a surviving copy. Treating a failed +// query as "the other side has it" is how a network blip becomes data loss. +func TestDedupRefusesWhenThePeerCannotBeQueried(t *testing.T) { + holder := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + unreachable := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + unreachable.stop() // peer down; its inventory is unknown, not empty + + err := newDedupTask().Execute(context.Background(), dedupParams(holder.address(), unreachable.address(), 0)) + + if holder.has(dedupTestVolumeID, 0) { + return // kept the shard while the peer's state is unknown + } + t.Fatalf("dedup deleted shard %d.0 while the peer was unreachable (deleted=%v, err=%v)", + dedupTestVolumeID, holder.deletedShards(), err) +} + +// Volume ids are allocated cluster-wide, but the inventory RPC is keyed by +// volume id alone, so a server holding the same number for a different +// collection answers "yes, I have that shard" to a question about this one. +// Accepting that would delete the last real copy on the strength of an +// unrelated volume. +func TestDedupRefusesWhenTheKeepNodeHoldsAnotherCollection(t *testing.T) { + holder := startFakeEcVolumeServer(t, dedupTestVolumeID, 0) + // Same volume id, different collection: a plausible-looking but wrong match. + impostor := startFakeEcVolumeServerInCollection(t, "morro-agudo-01", dedupTestVolumeID, 0) + + err := newDedupTask().Execute(context.Background(), dedupParams(holder.address(), impostor.address(), 0)) + + if holder.has(dedupTestVolumeID, 0) { + return // refused: the keep node's shard belongs to another collection + } + t.Fatalf("dedup deleted shard %d.0 after matching another collection's volume (deleted=%v, err=%v)", + dedupTestVolumeID, holder.deletedShards(), err) +} diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index 6d55b23cb..3999f6e00 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.go @@ -127,6 +127,7 @@ func Detection( TaskParams: &worker_pb.TaskParams_EcBalanceParams{ EcBalanceParams: &worker_pb.EcBalanceTaskParams{ DiskType: normalizedDiskType, + DedupKeepNode: m.KeepNode, TimeoutSeconds: 600, }, }, diff --git a/weed/worker/tasks/ec_balance/ec_balance_task.go b/weed/worker/tasks/ec_balance/ec_balance_task.go index 5fbb404c0..12cdaf3ab 100644 --- a/weed/worker/tasks/ec_balance/ec_balance_task.go +++ b/weed/worker/tasks/ec_balance/ec_balance_task.go @@ -90,7 +90,7 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam // For dedup, we only unmount+delete from source (no copy needed) if isDedupDelete { - return t.executeDedupDelete(ctx, params.VolumeId, sourceAddr, source.ShardIds) + return t.executeDedupDelete(ctx, params.VolumeId, sourceAddr, source.ShardIds, ecParams.GetDedupKeepNode()) } // Step 1: Copy shard to destination and mount @@ -127,8 +127,19 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam return nil } -// executeDedupDelete removes a duplicate shard without copying -func (t *ECBalanceTask) executeDedupDelete(ctx context.Context, volumeID uint32, sourceAddr pb.ServerAddress, shardIDs []uint32) error { +// executeDedupDelete removes a duplicate shard without copying. Because nothing +// is copied first, the only thing standing between this and data loss is that +// another node really holds the shard -- and the plan asserting so is not +// evidence. The topology can name a location that holds nothing (such a server +// answers "not found ec volume id" when asked for the file), and deleting on the +// strength of that removes the last copy while reporting success. So confirm the +// shard on the node the plan chose to keep, and keep this copy if that cannot be +// established. An unreachable peer is unknown, not confirmed. +func (t *ECBalanceTask) executeDedupDelete(ctx context.Context, volumeID uint32, sourceAddr pb.ServerAddress, shardIDs []uint32, keepNode string) error { + if err := t.verifyShardsOnKeepNode(ctx, volumeID, keepNode, shardIDs); err != nil { + return err + } + t.reportProgress(25.0, "Unmounting duplicate EC shard") if err := t.unmountShard(ctx, volumeID, sourceAddr, shardIDs); err != nil { return fmt.Errorf("unmount duplicate shard: %w", err) @@ -143,6 +154,19 @@ func (t *ECBalanceTask) executeDedupDelete(ctx context.Context, volumeID uint32, return nil } +// verifyShardsOnKeepNode confirms the node the plan wants to keep the shard on +// actually has every shard about to be deleted elsewhere, for this collection. +func (t *ECBalanceTask) verifyShardsOnKeepNode(ctx context.Context, volumeID uint32, keepNode string, shardIDs []uint32) error { + if keepNode == "" { + return fmt.Errorf("refusing dedup delete of volume %d shard(s) %v: no keep node recorded, so no surviving copy can be confirmed", volumeID, shardIDs) + } + if err := erasure_coding.VerifyShardsOnServer(ctx, t.collection, volumeID, + string(pb.ServerAddress(keepNode)), shardIDs, t.grpcDialOption); err != nil { + return fmt.Errorf("refusing dedup delete: %w", err) + } + return nil +} + // copyAndMountShard copies EC shard from source to destination and mounts it func (t *ECBalanceTask) copyAndMountShard(ctx context.Context, volumeID uint32, sourceAddr, targetAddr pb.ServerAddress, shardIDs []uint32, destDiskID uint32) error { return operation.WithVolumeServerClient(false, targetAddr, t.grpcDialOption, diff --git a/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go b/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go new file mode 100644 index 000000000..efc375209 --- /dev/null +++ b/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go @@ -0,0 +1,140 @@ +package ec_balance + +import ( + "context" + "fmt" + "net" + "sync" + "testing" + + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" +) + +// fakeEcVolumeServer is a volume server that only knows about EC shard +// inventory. It exists so the destructive half of a balance job can be driven +// in-process: what a real volume server has on disk is the thing the master's +// topology can be wrong about, and being wrong about it is what makes a delete +// unsafe. Holding that state here lets a test say "the topology claims this, +// the disk holds that" and watch what the job does. +type fakeEcVolumeServer struct { + volume_server_pb.UnimplementedVolumeServerServer + + mu sync.Mutex + // shards actually present on this server's disks, per volume. + shards map[uint32]map[uint32]bool + // collection the held volume belongs to. Reported back on the inventory so + // a caller can tell this server's volume N from another collection's N. + collection string + // deleted/unmounted record what the job asked for, so a test can assert on + // the request even when it was a no-op against the inventory. + deleted []string + unmounted []string + // grpcAddr is host:port; the code under test dials by address string. + grpcAddr string + stop func() +} + +func (f *fakeEcVolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + resp := &volume_server_pb.VolumeEcShardsInfoResponse{} + for sid := range f.shards[req.VolumeId] { + resp.EcShardInfos = append(resp.EcShardInfos, &volume_server_pb.EcShardInfo{ + VolumeId: req.VolumeId, + ShardId: sid, + Collection: f.collection, + }) + } + return resp, nil +} + +func (f *fakeEcVolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + f.deleted = append(f.deleted, fmt.Sprintf("%d.%d", req.VolumeId, sid)) + delete(f.shards[req.VolumeId], sid) + } + return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil +} + +func (f *fakeEcVolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + f.unmounted = append(f.unmounted, fmt.Sprintf("%d.%d", req.VolumeId, sid)) + } + return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil +} + +// has reports whether the shard is still on this server's disks. +func (f *fakeEcVolumeServer) has(volumeID, shardID uint32) bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.shards[volumeID][shardID] +} + +func (f *fakeEcVolumeServer) deletedShards() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.deleted...) +} + +// startFakeEcVolumeServer listens on a loopback port and serves the EC subset +// of the volume server API. present lists the shards this server really holds. +func startFakeEcVolumeServer(t *testing.T, volumeID uint32, present ...uint32) *fakeEcVolumeServer { + return startFakeEcVolumeServerInCollection(t, dedupTestCollection, volumeID, present...) +} + +// startFakeEcVolumeServerInCollection is the same, for a named collection, so a +// test can stand up two servers holding the same volume id in different ones. +func startFakeEcVolumeServerInCollection(t *testing.T, collection string, volumeID uint32, present ...uint32) *fakeEcVolumeServer { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + f := &fakeEcVolumeServer{ + shards: map[uint32]map[uint32]bool{volumeID: {}}, + collection: collection, + grpcAddr: listener.Addr().String(), + } + for _, sid := range present { + f.shards[volumeID][sid] = true + } + + server := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(server, f) + go func() { _ = server.Serve(listener) }() + f.stop = server.Stop + t.Cleanup(server.Stop) + + return f +} + +// address returns a node id in the "host:httpPort.grpcPort" form the cluster +// uses (e.g. mg01-s3-intelistor-15:8087.18087), so ToGrpcAddress resolves to +// this fake's listener. The http port is unused and only has to parse. +func (f *fakeEcVolumeServer) address() string { + host, port, err := net.SplitHostPort(f.grpcAddr) + if err != nil { + return f.grpcAddr + } + return fmt.Sprintf("%s:1.%s", host, port) +} + +// shardBitsOf is a small readability helper for asserting inventory. +func shardBitsOf(f *fakeEcVolumeServer, volumeID uint32) erasure_coding.ShardBits { + f.mu.Lock() + defer f.mu.Unlock() + var b erasure_coding.ShardBits + for sid := range f.shards[volumeID] { + b = b.Set(erasure_coding.ShardId(sid)) + } + return b +}