ec: refund the cleared leftover shards' slots in the encode source health check (#10903)

* erasure_coding: one home for the shard-count to volume-slots conversion

* ec: refund the cleared leftover shards' slots in the encode source health check

weed/ec/ec_encode.go

@@ -224,6 +224,55 @@ func volumeLocations(env *Env, volumeIds []needle.VolumeId) (map[needle.VolumeId
 	return res, nil
 }
 
+// collectSourceFreeVolumeCounts maps "volumeId-serverAddress" to the free
+// volume slots on the disk holding that volume, for the source health check.
+// Key by dn.Address so it matches wdclient.Location.Url: in deployments where
+// dn.Id is a short name (e.g. a Kubernetes StatefulSet pod name) while
+// dn.Address is a FQDN:port, keying by dn.Id would never match the location
+// Url during the lookup.
+//
+// The topology snapshot predates clearPreexistingEcShards, so shards of the
+// very volumes being encoded (leftovers of an interrupted run the sweep just
+// removed) may still be charged against a disk's FreeVolumeCount. Refund their
+// slots, or a retried encode fails the health check on capacity the sweep
+// already freed. A node the sweep skipped as unreachable keeps its charge: its
+// leftovers are still there.
+func collectSourceFreeVolumeCounts(topologyInfo *master_pb.TopologyInfo, volumeIds []needle.VolumeId, sweepSkippedNodes map[pb.ServerAddress]struct{}) map[string]int {
+	encoding := make(map[uint32]bool, len(volumeIds))
+	for _, vid := range volumeIds {
+		encoding[uint32(vid)] = true
+	}
+	freeVolumeCountMap := make(map[string]int) // key: volumeId-serverAddress
+	EachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		addr := dn.Address
+		if addr == "" {
+			addr = dn.Id // older nodes use ip:port as id
+		}
+		_, skipped := sweepSkippedNodes[pb.NewServerAddressFromDataNode(dn)]
+		for _, diskInfo := range dn.DiskInfos {
+			free := diskInfo.FreeVolumeCount
+			if !skipped {
+				var total, cleared int64
+				for _, ecInfo := range diskInfo.EcShardInfos {
+					n := int64(erasure_coding.GetShardCount(ecInfo))
+					total += n
+					if encoding[ecInfo.Id] {
+						cleared += n
+					}
+				}
+				if cleared > 0 {
+					free += erasure_coding.VolumeSlots(total) - erasure_coding.VolumeSlots(total-cleared)
+				}
+			}
+			for _, v := range diskInfo.VolumeInfos {
+				key := fmt.Sprintf("%d-%s", v.Id, addr)
+				freeVolumeCountMap[key] = int(free)
+			}
+		}
+	})
+	return freeVolumeCountMap
+}
+
 func doEcEncode(env *Env, writer io.Writer, volumeIdToCollection map[needle.VolumeId]string, volumeIds []needle.VolumeId, maxParallelization int, topologyInfo *master_pb.TopologyInfo) (skippedNodes map[pb.ServerAddress]struct{}, err error) {
 	if !env.isLocked() {
 		return nil, fmt.Errorf("lock is lost")
@@ -243,24 +292,7 @@ func doEcEncode(env *Env, writer io.Writer, volumeIdToCollection map[needle.Volu
 		return nil, fmt.Errorf("clear pre-existing ec shards before encoding: %w", err)
 	}
 
-	// Build a map of (volumeId, serverAddress) -> freeVolumeCount.
-	// Key by dn.Address so it matches wdclient.Location.Url. In deployments
-	// where dn.Id is a short name (e.g. Kubernetes StatefulSet pod name)
-	// while dn.Address is a FQDN:port, keying by dn.Id would never match the
-	// location Url during the health-check lookup below.
-	freeVolumeCountMap := make(map[string]int) // key: volumeId-serverAddress
-	EachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
-		addr := dn.Address
-		if addr == "" {
-			addr = dn.Id // older nodes use ip:port as id
-		}
-		for _, diskInfo := range dn.DiskInfos {
-			for _, v := range diskInfo.VolumeInfos {
-				key := fmt.Sprintf("%d-%s", v.Id, addr)
-				freeVolumeCountMap[key] = int(diskInfo.FreeVolumeCount)
-			}
-		}
-	})
+	freeVolumeCountMap := collectSourceFreeVolumeCounts(topologyInfo, volumeIds, skippedNodes)
 
 	// Filter replicas by free capacity BEFORE marking volumes readonly so that
 	// a failed health check does not strand volumes in readonly state.

weed/ec/ec_encode_test.go

@@ -475,3 +475,58 @@ func TestEcShardSummaryNamesTheShardIds(t *testing.T) {
 		"node2:8080=2 shards [1 2]",
 	}, ecShardSummaryByNode(byNode))
 }
+
+// A retried ec.encode clears the leftover shards of an interrupted run before
+// the source health check, but the topology snapshot predates that sweep: the
+// leftovers still depress FreeVolumeCount on the source disk. The counts must
+// refund those slots, or the retry refuses the very volume it just cleaned up
+// after. Numbers mirror the interruption-matrix failure: 8 slots, 4 volumes,
+// 8 shards of other EC volumes, plus 14 leftover shards of volume 3, so the
+// master charges ceil(22/10)=3 slots and reports free=1.
+func TestCollectSourceFreeVolumeCountsRefundsClearedShards(t *testing.T) {
+	newTopo := func() *master_pb.TopologyInfo {
+		return &master_pb.TopologyInfo{
+			DataCenterInfos: []*master_pb.DataCenterInfo{{
+				Id: "dc1",
+				RackInfos: []*master_pb.RackInfo{{
+					Id: "rack0",
+					DataNodeInfos: []*master_pb.DataNodeInfo{{
+						Id:      "127.0.0.1:8110",
+						Address: "127.0.0.1:8110",
+						DiskInfos: map[string]*master_pb.DiskInfo{
+							"hdd": {
+								Type:            "hdd",
+								MaxVolumeCount:  8,
+								VolumeCount:     4,
+								FreeVolumeCount: 1,
+								VolumeInfos: []*master_pb.VolumeInformationMessage{
+									{Id: 3}, {Id: 6},
+								},
+								EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
+									{Id: 1, EcIndexBits: 0b00000010001, DiskId: 0}, // shards 0,4
+									{Id: 1, EcIndexBits: 0b10000000100, DiskId: 1}, // shards 2,10
+									{Id: 2, EcIndexBits: 0b00000010001, DiskId: 0},
+									{Id: 2, EcIndexBits: 0b10000000100, DiskId: 1},
+									{Id: 3, EcIndexBits: 0b11111111111111, DiskId: 0}, // interrupted run's leftovers
+								},
+							},
+						},
+					}},
+				}},
+			}},
+		}
+	}
+
+	counts := collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{3}, nil)
+	assert.Equal(t, 3, counts["3-127.0.0.1:8110"],
+		"the cleared leftovers' ceil(22/10)-ceil(8/10)=2 slots must be refunded")
+
+	counts = collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{6}, nil)
+	assert.Equal(t, 1, counts["6-127.0.0.1:8110"],
+		"no refund when the encoded volume has no leftover shards")
+
+	skipped := map[pb.ServerAddress]struct{}{"127.0.0.1:8110": {}}
+	counts = collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{3}, skipped)
+	assert.Equal(t, 1, counts["3-127.0.0.1:8110"],
+		"no refund on a node the sweep skipped as unreachable")
+}

weed/storage/erasure_coding/ec_shards_info.go

@@ -104,6 +104,13 @@ func GetShardCount(vi *master_pb.VolumeEcShardInformationMessage) int {
 	return ShardBits(vi.EcIndexBits).Count()
 }
 
+// VolumeSlots converts an EC shard count to the number of volume slots those
+// shards occupy: every DataShardsCount shards hold one volume's worth of data,
+// rounded up.
+func VolumeSlots(ecShardCount int64) int64 {
+	return (ecShardCount + DataShardsCount - 1) / DataShardsCount
+}
+
 // EcShardsTotalSize returns the sum of all shard sizes (data + parity) in
 // the message. Walks vi.ShardSizes directly rather than materializing a
 // ShardsInfo, which is significantly cheaper for callers that only need the

weed/storage/store.go

@@ -467,8 +467,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		effectiveMaxCount := location.MaxVolumeCount
 		if location.isDiskSpaceLow.Load() {
 			usedSlots := int32(location.LocalVolumesLen())
-			ecShardCount := location.EcShardCount()
-			usedSlots += int32((ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount)
+			usedSlots += int32(erasure_coding.VolumeSlots(int64(location.EcShardCount())))
 			effectiveMaxCount = usedSlots
 		}
 		if effectiveMaxCount < 0 {

weed/topology/disk.go

@@ -50,12 +50,6 @@ type Disk struct {
 // from lingering forever.
 const volumeRemovalGracePeriod = 10 * time.Second
 
-// ecShardSlots returns the number of volume slots consumed by the given
-// number of EC shards, rounded up to whole-volume equivalents.
-func ecShardSlots(ecShardCount int64) int64 {
-	return (ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount
-}
-
 func NewDisk(diskType string) *Disk {
 	s := &Disk{}
 	s.id = NodeId(diskType)
@@ -106,7 +100,7 @@ func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo {
 		m := &master_pb.DiskInfo{
 			VolumeCount:       usage.volumeCount,
 			MaxVolumeCount:    usage.maxVolumeCount,
-			FreeVolumeCount:   usage.maxVolumeCount - (usage.volumeCount - usage.remoteVolumeCount) - ecShardSlots(usage.ecShardCount),
+			FreeVolumeCount:   usage.maxVolumeCount - (usage.volumeCount - usage.remoteVolumeCount) - erasure_coding.VolumeSlots(usage.ecShardCount),
 			ActiveVolumeCount: usage.activeVolumeCount,
 			RemoteVolumeCount: usage.remoteVolumeCount,
 			DiskTotalBytes:    uint64(max(0, usage.diskTotalBytes)),
@@ -174,7 +168,7 @@ func (a *DiskUsageCounts) snapshot() DiskUsageCounts {
 
 func (a *DiskUsageCounts) FreeSpace() int64 {
 	u := a.snapshot()
-	return u.maxVolumeCount + u.remoteVolumeCount - u.volumeCount - ecShardSlots(u.ecShardCount)
+	return u.maxVolumeCount + u.remoteVolumeCount - u.volumeCount - erasure_coding.VolumeSlots(u.ecShardCount)
 }
 
 func (du *DiskUsages) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
@@ -452,7 +446,7 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo {
 		Type:              string(d.Id()),
 		VolumeCount:       diskUsage.volumeCount,
 		MaxVolumeCount:    diskUsage.maxVolumeCount,
-		FreeVolumeCount:   diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - ecShardSlots(diskUsage.ecShardCount),
+		FreeVolumeCount:   diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - erasure_coding.VolumeSlots(diskUsage.ecShardCount),
 		ActiveVolumeCount: diskUsage.activeVolumeCount,
 		RemoteVolumeCount: diskUsage.remoteVolumeCount,
 		DiskId:            diskId,

weed/topology/node.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
@@ -308,7 +309,7 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
 func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 {
 	t := n.getOrCreateDisk(option.DiskType)
 	freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount)
-	freeVolumeSlotCount -= ecShardSlots(atomic.LoadInt64(&t.ecShardCount))
+	freeVolumeSlotCount -= erasure_coding.VolumeSlots(atomic.LoadInt64(&t.ecShardCount))
 	return freeVolumeSlotCount
 }