feat(fibre/validator): add Select for stake-weighted validator selection

Add Select function that returns validators shuffled by stake for
load balancing during downloads. Higher-stake validators are more
likely to appear earlier, distributing load proportionally.

Author: Wondertan

fibre/validator/set.go

@@ -96,6 +96,78 @@ func (s Set) Assign(commitment rsema1d.Commitment, totalRows, originalRows, minR
 	return shardMap
 }
 
+// Select returns validators to download shards from, shuffled by stake for load balancing.
+// Returns all validators and the minimum count needed for reconstruction (covering livenessThreshold stake).
+func (s Set) Select(originalRows, minRows int, livenessThreshold cmtmath.Fraction) (validators []*core.Validator, minRequired int) {
+	if len(s.Validators) == 0 {
+		return nil, 0
+	}
+
+	validators = make([]*core.Validator, len(s.Validators))
+	copy(validators, s.Validators)
+
+	// minStakeFraction is the minimum contribution per validator for unique decodability
+	// e.g., 148 / (4096 * 3) ≈ 1.2% for livenessThreshold=1/3
+	totalDistributedRows := originalRows * int(livenessThreshold.Denominator) / int(livenessThreshold.Numerator)
+	minStakeFraction := float64(minRows) / float64(totalDistributedRows)
+
+	// find last non-overlapping validator using effective stake (actual stake floored by minStakeFraction)
+	totalStake := float64(s.TotalVotingPower())
+	accumulated := 0.0
+	splitIdx := len(validators)
+	for i, v := range validators {
+		accumulated += max(float64(v.VotingPower)/totalStake, minStakeFraction)
+		if accumulated >= 1.0 {
+			splitIdx = i + 1
+			break
+		}
+	}
+
+	// shuffle each group by stake for load balancing
+	// NOTE: doesn't require cryptographic randomness
+	rng := rand.New(rand.NewPCG(rand.Uint64(), rand.Uint64()))
+	shuffleByStake(validators[:splitIdx], rng)
+	shuffleByStake(validators[splitIdx:], rng)
+
+	// count validators needed to cover livenessThreshold stake
+	livenessStake := s.TotalVotingPower() * int64(livenessThreshold.Numerator) / int64(livenessThreshold.Denominator)
+	coveredStake := int64(0)
+	for _, v := range validators[:splitIdx] {
+		minRequired++
+		coveredStake += v.VotingPower
+		if coveredStake >= livenessStake {
+			break
+		}
+	}
+
+	return validators, minRequired
+}
+
+// shuffleByStake shuffles validators in-place using stake-weighted random selection.
+// Validators with higher voting power are more likely to appear earlier.
+func shuffleByStake(validators []*core.Validator, rng *rand.Rand) {
+	for i := range len(validators) - 1 {
+		// calculate total weight of remaining validators
+		var totalWeight int64
+		for j := i; j < len(validators); j++ {
+			totalWeight += validators[j].VotingPower
+		}
+
+		// pick random point in weight space
+		point := rng.Int64N(totalWeight)
+
+		// find and swap the selected validator
+		var cumul int64
+		for j := i; j < len(validators); j++ {
+			cumul += validators[j].VotingPower
+			if point < cumul {
+				validators[i], validators[j] = validators[j], validators[i]
+				break
+			}
+		}
+	}
+}
+
 // Verify checks if all given row indices are assigned to [core.Validator].
 // Returns error if validator is not in the map, count doesn't match, or any row is not assigned.
 // This method builds a temporary set for O(r + n) complexity instead of O(n × r).

fibre/validator/set_bench_test.go

@@ -0,0 +1,59 @@
+package validator_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/celestiaorg/celestia-app/v6/fibre/validator"
+	"github.com/cometbft/cometbft/crypto/ed25519"
+	core "github.com/cometbft/cometbft/types"
+)
+
+// BenchmarkSet_Assign measures row assignment performance.
+// Dominated by Fisher-Yates shuffle of 16384 rows, constant across validator counts.
+//
+// Results (AMD Ryzen 9 7940HS):
+//
+//	10_validators     135 µs/op   235 kB/op    18 allocs/op
+//	50_validators     136 µs/op   239 kB/op    62 allocs/op
+//	100_validators    135 µs/op   244 kB/op   114 allocs/op
+func BenchmarkSet_Assign(b *testing.B) {
+	for _, n := range []int{10, 50, 100} {
+		b.Run(fmt.Sprintf("%d_validators", n), func(b *testing.B) {
+			valSet := makeBenchValidatorSet(n)
+			for b.Loop() {
+				_ = valSet.Assign(testCommitment, 16384, 4096, 16, testLivenessThreshold)
+			}
+		})
+	}
+}
+
+// BenchmarkSet_Select measures validator selection performance.
+// Scales O(n²) due to stake-weighted shuffle but remains negligible for typical validator counts.
+//
+// Results (AMD Ryzen 9 7940HS):
+//
+//	10_validators     196 ns/op    96 B/op     2 allocs/op
+//	50_validators     1.5 µs/op   432 B/op     2 allocs/op
+//	100_validators    3.7 µs/op   912 B/op     2 allocs/op
+func BenchmarkSet_Select(b *testing.B) {
+	for _, n := range []int{10, 50, 100} {
+		b.Run(fmt.Sprintf("%d_validators", n), func(b *testing.B) {
+			valSet := makeBenchValidatorSet(n)
+			for b.Loop() {
+				_, _ = valSet.Select(testOriginalRows, testMinRows, testLivenessThreshold)
+			}
+		})
+	}
+}
+
+func makeBenchValidatorSet(n int) validator.Set {
+	validators := make([]*core.Validator, n)
+	for i := range n {
+		validators[i] = core.NewValidator(ed25519.GenPrivKey().PubKey(), 1)
+	}
+	return validator.Set{
+		ValidatorSet: core.NewValidatorSet(validators),
+		Height:       100,
+	}
+}