fix(ec): pass per-volume data-shard count to the parity-shard split (#9781)

* fix(ec): pass per-volume data-shard count to the parity-shard split

ShardsInfo.DeleteParityShards/MinusParityShards looped ids 10..13, assuming
the fixed 10+4 layout. For a non-default ratio this splits data vs parity
wrong — a wide ratio (12+4, 16+6) drops real data ids >= 10, which breaks
ec.decode. They now take a dataShards argument (<= 0 falls back to
DataShardsCount) and clear ids dataShards..MaxShardCount. ec.decode threads
the data-shard count from collectEcNodeShardsInfo to both split call sites,
and admin LogicalSize passes DataShardsCount.

Also: EC cleanup now sets an explicit per-disk storage impact
(-len(ShardIds)) instead of falling back to the TotalShardsCount constant,
so freed-capacity accounting matches the shards actually removed.

OSS is always 10+4, so behavior is unchanged here; this keeps the split
ratio-correct and the API aligned with the enterprise per-volume override.
Adds parity-split ratio tests.

* ec: clear parity shards in one locked pass

Address review: DeleteParityShards looped si.Delete, taking the lock once per
id. shards is sorted by Id and shardBits is a bitmap, so mask off the high
bits and truncate the sorted slice at the first parity id (binary search) under
a single lock. Preserves the dataShards<=0 -> DataShardsCount default.

weed/admin/dash/admin_server.go

@@ -1778,7 +1778,7 @@ func collectCollectionStats(topologyInfo *master_pb.TopologyInfo) map[string]col
 						shards := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(ecShardInfo)
 						data := collectionMap[collection]
 						data.PhysicalSize += int64(shards.TotalSize())
-						data.LogicalSize += int64(shards.MinusParityShards().TotalSize())
+						data.LogicalSize += int64(shards.MinusParityShards(erasure_coding.DataShardsCount).TotalSize())
 						collectionMap[collection] = data
 
 						// fileCount is volume-wide (same .ecx on every shard

weed/shell/command_ec_decode.go

@@ -115,7 +115,7 @@ func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collec
 	}
 
 	// find volume location
-	nodeToEcShardsInfo := collectEcNodeShardsInfo(topoInfo, vid, diskType)
+	nodeToEcShardsInfo, dataShards := collectEcNodeShardsInfo(topoInfo, vid, diskType)
 
 	fmt.Printf("ec volume %d shard locations: %+v\n", vid, nodeToEcShardsInfo)
 
@@ -148,7 +148,7 @@ func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collec
 	}
 
 	// collect ec shards to the server with most space
-	targetNodeLocation, err := collectEcShards(commandEnv, nodeToEcShardsInfo, collection, vid, eligibleTargets)
+	targetNodeLocation, err := collectEcShards(commandEnv, nodeToEcShardsInfo, collection, vid, eligibleTargets, dataShards)
 	if err != nil {
 		return fmt.Errorf("collectEcShards for volume %d: %v", vid, err)
 	}
@@ -281,7 +281,7 @@ func generateNormalVolume(grpcDialOption grpc.DialOption, vid needle.VolumeId, c
 
 }
 
-func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddress]*erasure_coding.ShardsInfo, collection string, vid needle.VolumeId, eligibleTargets map[pb.ServerAddress]struct{}) (targetNodeLocation pb.ServerAddress, err error) {
+func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddress]*erasure_coding.ShardsInfo, collection string, vid needle.VolumeId, eligibleTargets map[pb.ServerAddress]struct{}, dataShards int) (targetNodeLocation pb.ServerAddress, err error) {
 
 	maxShardCount := -1
 	existingShardsInfo := erasure_coding.NewShardsInfo()
@@ -291,7 +291,7 @@ func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddre
 				continue
 			}
 		}
-		toBeCopiedShardCount := si.MinusParityShards().Count()
+		toBeCopiedShardCount := si.MinusParityShards(dataShards).Count()
 		if toBeCopiedShardCount > maxShardCount {
 			maxShardCount = toBeCopiedShardCount
 			targetNodeLocation = loc
@@ -310,7 +310,7 @@ func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddre
 			continue
 		}
 
-		needToCopyShardsInfo := si.Minus(existingShardsInfo).MinusParityShards()
+		needToCopyShardsInfo := si.Minus(existingShardsInfo).MinusParityShards(dataShards)
 
 		err = operation.WithVolumeServerClient(false, targetNodeLocation, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 
@@ -401,7 +401,7 @@ func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern strin
 	return
 }
 
-func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) map[pb.ServerAddress]*erasure_coding.ShardsInfo {
+func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) (map[pb.ServerAddress]*erasure_coding.ShardsInfo, int) {
 	res := make(map[pb.ServerAddress]*erasure_coding.ShardsInfo)
 	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
 		if diskInfo, found := dn.DiskInfos[string(diskType)]; found {
@@ -423,7 +423,8 @@ func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.Volume
 		}
 	})
 
-	return res
+	// OSS is always 10+4; the per-volume ratio override lives in the enterprise build.
+	return res, erasure_coding.DataShardsCount
 }
 
 type decodeDiskUsageState struct {

weed/shell/command_ec_encode.go

@@ -359,7 +359,7 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum
 
 		lastErr = nil
 		for _, vid := range volumeIds {
-			nodeShards := collectEcNodeShardsInfo(topoInfo, vid, diskType)
+			nodeShards, _ := collectEcNodeShardsInfo(topoInfo, vid, diskType)
 
 			var union erasure_coding.ShardBits
 			for _, info := range nodeShards {

weed/storage/erasure_coding/ec_parity_shards_ratio_test.go

@@ -0,0 +1,49 @@
+package erasure_coding
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestMinusParityShardsRatio covers the data/parity split for the default 10+4
+// layout and other ratios. The pre-fix code hardcoded ids 10..13 as parity,
+// which for a wide ratio (12+4, 16+6) DROPPED real data ids >= 10 and for a
+// narrow ratio (9+3) kept parity id 9. The method now takes the data-shard
+// count (<= 0 falls back to DataShardsCount).
+func TestMinusParityShardsRatio(t *testing.T) {
+	all := func(n int) []ShardId {
+		ids := make([]ShardId, n)
+		for i := range ids {
+			ids[i] = ShardId(i)
+		}
+		return ids
+	}
+	tests := []struct {
+		name          string
+		dataShards    int
+		inputShards   []ShardId
+		expectedCount int
+		expectedIds   []ShardId
+	}{
+		{"default 10+4 all present", DataShardsCount, all(14), 10, all(10)},
+		{"zero ratio falls back to default", 0, all(14), 10, all(10)},
+		{"9+3 all present", 9, all(12), 9, all(9)},
+		{"9+3 drops parity id 9", 9, []ShardId{0, 1, 2, 9, 10, 11}, 3, []ShardId{0, 1, 2}},
+		{"12+4 keeps data ids >= 10", 12, all(16), 12, all(12)},
+		{"16+6 keeps data ids >= 10", 16, all(22), 16, all(16)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			si := NewShardsInfo()
+			for _, id := range tt.inputShards {
+				si.Set(NewShardInfo(id, ShardSize(1000)))
+			}
+			result := si.MinusParityShards(tt.dataShards)
+			assert.Equal(t, tt.expectedCount, result.Count(), "count for ratio %d", tt.dataShards)
+			for _, id := range tt.expectedIds {
+				assert.True(t, result.Has(id), "data shard %d must remain for ratio %d", id, tt.dataShards)
+			}
+		})
+	}
+}

weed/storage/erasure_coding/ec_shards_info.go

@@ -338,17 +338,37 @@ func (si *ShardsInfo) Copy() *ShardsInfo {
 	}
 }
 
-// DeleteParityShards removes parity shards from a ShardInfo.
-func (si *ShardsInfo) DeleteParityShards() {
-	for id := DataShardsCount; id < TotalShardsCount; id++ {
-		si.Delete(ShardId(id))
+// DeleteParityShards removes parity shards (those with id >= dataShards) from
+// a ShardInfo. dataShards is the volume's data-shard count; passing <= 0 falls
+// back to DataShardsCount (the fixed 10+4 layout). The upper bound is
+// MaxShardCount, not TotalShardsCount, so a custom ratio's high parity ids
+// (e.g. 16+6 reaches id 21) are cleared too; Delete no-ops on absent ids.
+func (si *ShardsInfo) DeleteParityShards(dataShards int) {
+	if dataShards <= 0 {
+		dataShards = DataShardsCount
 	}
+	if dataShards >= MaxShardCount {
+		return // every id is a data shard; nothing to remove
+	}
+	si.mu.Lock()
+	defer si.mu.Unlock()
+
+	// Parity ids are >= dataShards. shards stays sorted by Id and shardBits is
+	// a bitmap, so clear them in one locked pass instead of a per-id Delete:
+	// mask off the high bits, then truncate the sorted slice at the first
+	// parity id via binary search.
+	si.shardBits &= ShardBits((uint32(1) << uint(dataShards)) - 1)
+	idx := sort.Search(len(si.shards), func(i int) bool {
+		return si.shards[i].Id >= ShardId(dataShards)
+	})
+	si.shards = si.shards[:idx]
 }
 
-// MinusParityShards creates a ShardInfo copy, but with parity shards removed.
-func (si *ShardsInfo) MinusParityShards() *ShardsInfo {
+// MinusParityShards creates a ShardInfo copy with parity shards removed for the
+// given data-shard count (<= 0 falls back to DataShardsCount).
+func (si *ShardsInfo) MinusParityShards(dataShards int) *ShardsInfo {
 	result := si.Copy()
-	result.DeleteParityShards()
+	result.DeleteParityShards(dataShards)
 	return result
 }
 

weed/storage/erasure_coding/ec_shards_info_test.go

@@ -243,7 +243,7 @@ func TestShardsInfo_DeleteParityShards(t *testing.T) {
 		si.Set(ShardInfo{Id: ShardId(i), Size: ShardSize((i + 1) * 1000)})
 	}
 
-	si.DeleteParityShards()
+	si.DeleteParityShards(DataShardsCount)
 
 	// Verify only data shards remain
 	for i := 0; i < DataShardsCount; i++ {

weed/storage/erasure_coding/ec_volume_info_test.go

@@ -13,7 +13,7 @@ func TestShardsInfoDeleteParityShards(t *testing.T) {
 	for _, id := range erasure_coding.AllShardIds() {
 		si.Set(erasure_coding.ShardInfo{Id: id, Size: 123})
 	}
-	si.DeleteParityShards()
+	si.DeleteParityShards(erasure_coding.DataShardsCount)
 
 	if got, want := si.String(), "0:123 B 1:123 B 2:123 B 3:123 B 4:123 B 5:123 B 6:123 B 7:123 B 8:123 B 9:123 B"; got != want {
 		t.Errorf("expected %q, got %q", want, got)

weed/worker/tasks/erasure_coding/detection.go

@@ -307,13 +307,19 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 				for _, shard := range existingECShards {
 					key := fmt.Sprintf("%s:%d", shard.ServerID, shard.DiskID)
 					if !duplicateCheck[key] { // Avoid duplicates if EC shards are on same disk as volume replicas
+						shardIds := append([]uint32(nil), shard.ShardIds...)
+						// Free exactly the shards on this disk. Without an explicit
+						// impact the cleanup falls back to CalculateECShardCleanupImpact,
+						// which credits TotalShardsCount (14) regardless of ratio.
+						cleanupImpact := topology.StorageSlotChange{ShardSlots: -int32(len(shardIds))}
 						sources = append(sources, topology.TaskSourceSpec{
-							ServerID:    shard.ServerID,
-							DiskID:      shard.DiskID,
-							DataCenter:  shard.DataCenter,
-							Rack:        shard.Rack,
-							CleanupType: topology.CleanupECShards,
-							ShardIds:    append([]uint32(nil), shard.ShardIds...),
+							ServerID:      shard.ServerID,
+							DiskID:        shard.DiskID,
+							DataCenter:    shard.DataCenter,
+							Rack:          shard.Rack,
+							CleanupType:   topology.CleanupECShards,
+							ShardIds:      shardIds,
+							StorageImpact: &cleanupImpact,
 						})
 						duplicateCheck[key] = true
 					}