ec: confirm a surviving copy before deleting a duplicate EC shard (#10719)

* ec: confirm a surviving copy before deleting a duplicate EC shard

The dedup phase of EC balancing removes a shard it believes exists elsewhere.
It copies nothing first, so the shard surviving on another node is the only
thing that makes the delete safe -- and it took the plan's word for that.

The plan is built from the master's topology, which can name a location that
holds nothing: such a server answers "CopyFile not found ec volume id N" when
something later tries to read the shard there. A shard listed on a phantom
location and on a real one looks duplicated, so dedup deletes one of them. When
it picks the real one the last copy is gone, and the job reports success -- the
loss only surfaces later, as a rebuild that cannot assemble enough shards.

The move phase already refuses to work on trust: it verifies the shard
registered on the destination before removing the source. Dedup now holds to
the same standard. The planner records which node it chose to keep, and both
executors -- the worker task and the shell's ec.balance -- confirm that node
really holds the shard before deleting. A keep node that cannot be queried is
unknown rather than confirmed, and blocks the delete.

Tests drive the destructive path against an in-process volume server that
tracks what is actually on disk separately from what the plan claims, which is
the distinction the bug turns on. Without the guard, two of them fail by
deleting the only copy and returning success.

* ec: check the collection and bound the wait when confirming a survivor

Two gaps in the dedup survivor check.

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 deletes the last real copy on the
strength of an unrelated volume. The response already carries the collection,
so verify against it rather than widening the RPC.

The shell path also queried on a background context, so a keep node that
accepts the connection but never answers would hang the whole balance run
instead of reporting that the survivor could not be confirmed. Bound it.

The check moves into VerifyShardsOnServer next to the existing helper, shared
by both executors, so the two paths cannot drift.

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

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" +

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

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,
 			})
 		}
 	}

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
+}

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)
+}

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,
 					},
 				},

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,

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
+}