feat(fibre/validator): stake-based assign

Author: Wondertan

fibre/blob.go

@@ -91,6 +91,11 @@ func NewBlobConfigFromParams(blobVersion uint8, p ProtocolParams) BlobConfig {
 	}
 }
 
+// TotalRows returns the total number of rows (OriginalRows + ParityRows).
+func (c BlobConfig) TotalRows() int {
+	return c.OriginalRows + c.ParityRows
+}
+
 // UploadSize calculates size of blob data with padding and w/o parity.
 // This is the size included in the [PaymentPromise] and the one actually paid for.
 func (c BlobConfig) UploadSize(dataLen int) int {

fibre/client.go

@@ -35,8 +35,11 @@ type ClientConfig struct {
 	// ChainID is the chain identifier for domain separation in [PaymentPromise] signatures.
 	ChainID string
 
-	// RowsPerShard computes the number of rows per shard given the total number of shards.
-	RowsPerShard func(totalShards int) int
+	// MinRowsPerValidator computes the minimum number of rows each validator must receive
+	// given the total number of validators, for unique decodability security.
+	MinRowsPerValidator func(totalValidators int) int
+	// LivenessThreshold is the fraction of stake needed to cause a liveness failure (typically 1/3).
+	LivenessThreshold cmtmath.Fraction
 	// MaxMessageSize is the maximum gRPC message size for upload requests.
 	MaxMessageSize int
 
@@ -76,7 +79,8 @@ func NewClientConfigFromParams(p ProtocolParams) ClientConfig {
 	return ClientConfig{
 		DefaultKeyName:              DefaultKeyName,
 		ChainID:                     "celestia",
-		RowsPerShard:                p.RowsPerShard,
+		MinRowsPerValidator:         p.MinRowsPerValidator,
+		LivenessThreshold:           p.LivenessThreshold,
 		MaxMessageSize:              p.MaxMessageSize(p.MaxValidatorCount),
 		UploadTargetVotingPower:     p.SafetyThreshold,
 		UploadTargetSignaturesCount: p.SafetyThreshold,

fibre/client_upload.go

@@ -67,7 +67,7 @@ func (c *Client) Upload(ctx context.Context, ns share.Namespace, blob *Blob) (re
 	))
 
 	// 2) assign shards to validators
-	shardMap := valSet.Assign(rsema1d.Commitment(blob.Commitment()), c.cfg.RowsPerShard(len(valSet.Validators)))
+	shardMap := valSet.Assign(rsema1d.Commitment(blob.Commitment()), blob.Config().TotalRows(), blob.Config().OriginalRows, c.cfg.MinRowsPerValidator(len(valSet.Validators)), c.cfg.LivenessThreshold)
 	span.AddEvent("shards_assigned")
 
 	validatorSignBytes, err := promise.SignBytesValidator()

fibre/protocol_params.go

@@ -113,6 +113,13 @@ func (p ProtocolParams) RowsPerShard(totalShards int) int {
 	return max(uniqueDecodeSamples, reconstructionSamples)
 }
 
+// MinRowsPerValidator returns the minimum number of rows each validator must receive
+// for unique decodability security, regardless of their stake percentage.
+// This is equivalent to RowsPerShard and serves as the floor in stake-aware assignment.
+func (p ProtocolParams) MinRowsPerValidator(totalValidators int) int {
+	return p.RowsPerShard(totalValidators)
+}
+
 // ShardsForReconstruction returns the minimum number of shards
 // needed to successfully reconstruct the original data.
 // Returns at least 1.

fibre/server.go

@@ -9,6 +9,7 @@ import (
 	"github.com/celestiaorg/celestia-app/v6/fibre/validator"
 	"github.com/celestiaorg/celestia-app/v6/x/fibre/types"
 	"github.com/cometbft/cometbft/crypto"
+	cmtmath "github.com/cometbft/cometbft/libs/math"
 	coregrpc "github.com/cometbft/cometbft/rpc/grpc"
 	core "github.com/cometbft/cometbft/types"
 	"github.com/cosmos/gogoproto/grpc"
@@ -24,8 +25,11 @@ type ServerConfig struct {
 	BlobConfig
 	StoreConfig
 
-	// RowsPerShard computes the number of rows per shard given the total number of shards.
-	RowsPerShard func(totalShards int) int
+	// MinRowsPerValidator computes the minimum number of rows each validator must receive
+	// given the total number of validators, for unique decodability security.
+	MinRowsPerValidator func(totalValidators int) int
+	// LivenessThreshold is the fraction of stake needed for reconstruction (typically 1/3).
+	LivenessThreshold cmtmath.Fraction
 	// MaxMessageSize is the maximum gRPC message size for upload requests.
 	MaxMessageSize int
 
@@ -46,11 +50,12 @@ func DefaultServerConfig() ServerConfig {
 // Use this when you need a config with non-default protocol parameters (e.g., for testing).
 func NewServerConfigFromParams(p ProtocolParams) ServerConfig {
 	return ServerConfig{
-		ChainID:        "celestia",
-		BlobConfig:     DefaultBlobConfigV0(), // currently hardcode support for version zero only
-		StoreConfig:    DefaultStoreConfig(),
-		RowsPerShard:   p.RowsPerShard,
-		MaxMessageSize: p.MaxMessageSize(p.MaxValidatorCount),
+		ChainID:             "celestia",
+		BlobConfig:          DefaultBlobConfigV0(), // currently hardcode support for version zero only
+		StoreConfig:         DefaultStoreConfig(),
+		MinRowsPerValidator: p.MinRowsPerValidator,
+		LivenessThreshold:   p.LivenessThreshold,
+		MaxMessageSize:      p.MaxMessageSize(p.MaxValidatorCount),
 	}
 }
 

fibre/server_upload.go

@@ -157,7 +157,7 @@ func (s *Server) verifyAssignment(ctx context.Context, promise *PaymentPromise,
 	for i, row := range shard.GetRows() {
 		rowIndices[i] = row.Index
 	}
-	shardMap := valSet.Assign(rsema1d.Commitment(promise.Commitment), s.cfg.RowsPerShard(len(valSet.Validators)))
+	shardMap := valSet.Assign(rsema1d.Commitment(promise.Commitment), s.cfg.TotalRows(), s.cfg.OriginalRows, s.cfg.MinRowsPerValidator(len(valSet.Validators)), s.cfg.LivenessThreshold)
 	if err := shardMap.Verify(ourValidator, rowIndices); err != nil {
 		return err
 	}

fibre/server_upload_test.go

@@ -84,7 +84,7 @@ func TestServerUploadShard(t *testing.T) {
 				// get commitment from the request (it's already a byte slice)
 				var commitment rsema1d.Commitment
 				copy(commitment[:], req.Promise.Commitment)
-				shardMap := valSet.Assign(commitment, cfg.RowsPerShard(len(valSet.Validators)))
+				shardMap := valSet.Assign(commitment, cfg.TotalRows(), cfg.OriginalRows, cfg.MinRowsPerValidator(len(valSet.Validators)), cfg.LivenessThreshold)
 				for val, indices := range shardMap {
 					if val.Address.String() != serverValidator.Address.String() && len(indices) > 0 {
 						req.Shard.Rows[0].Index = uint32(indices[0])
@@ -227,7 +227,8 @@ func makeTestRequest(
 	signPromise(promisePb)
 
 	// get row assignment for server validator
-	shardMap := valSet.Assign(rsema1d.Commitment(blob.Commitment()), fibre.DefaultProtocolParams.RowsPerShard(len(valSet.Validators)))
+	cfg := fibre.DefaultServerConfig()
+	shardMap := valSet.Assign(rsema1d.Commitment(blob.Commitment()), cfg.TotalRows(), cfg.OriginalRows, cfg.MinRowsPerValidator(len(valSet.Validators)), cfg.LivenessThreshold)
 	rowIndices := shardMap[serverValidator]
 	require.NotEmpty(t, rowIndices, "server validator has no rows assigned")
 

fibre/validator/set.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/celestiaorg/rsema1d"
 	"github.com/cometbft/cometbft/crypto"
+	cmtmath "github.com/cometbft/cometbft/libs/math"
 	core "github.com/cometbft/cometbft/types"
 )
 
@@ -33,25 +34,42 @@ type ShardMap map[*core.Validator][]int
 // Assign returns a ShardMap containing all validators and their assigned row indices
 // for the given commitment.
 //
-// The rowsPerValidator parameter specifies how many rows each validator receives.
-// Total rows distributed = rowsPerValidator * len(validators).
+// Rows are distributed based on the relation: originalRows / rows = livenessThreshold / stake%
+// This means 33% stake should have originalRows (4096), so each validator gets:
+// rows = ceil(originalRows * stake% / livenessThreshold)
 //
-// It uses a chacha8 RNG with the commitment as the seed to shuffle the row indices
-// using the Fisher-Yates algorithm.
-func (s Set) Assign(commitment rsema1d.Commitment, rowsPerValidator int) ShardMap {
-	if len(s.Validators) == 0 || rowsPerValidator == 0 {
+// The minRows parameter ensures every validator receives at least that many rows
+// for unique decodability security, even if their proportional share would be less.
+//
+// When the sum of assigned rows exceeds totalRows (due to minRows floor guarantees),
+// row indices wrap around using modulo arithmetic. This means the same row may be
+// assigned to multiple validators, ensuring all validators receive their required
+// minimum while maintaining deterministic assignment.
+//
+// It uses a ChaCha8 RNG seeded with the commitment to shuffle the row indices
+// using the Fisher-Yates algorithm, ensuring deterministic and uniform distribution.
+func (s Set) Assign(commitment rsema1d.Commitment, totalRows, originalRows, minRows int, livenessThreshold cmtmath.Fraction) ShardMap {
+	if len(s.Validators) == 0 || totalRows == 0 || minRows == 0 {
 		return make(ShardMap)
 	}
 
-	totalRows := rowsPerValidator * len(s.Validators)
+	// rows = ceil(originalRows * stake% / livenessThreshold)
+	//      = ceil(originalRows * votingPower * denominator / (totalVotingPower * numerator))
+	rowsPerValidator := make([]int, len(s.Validators))
+	for i, v := range s.Validators {
+		num := int64(originalRows) * v.VotingPower * int64(livenessThreshold.Denominator)
+		den := s.TotalVotingPower() * int64(livenessThreshold.Numerator)
+		rows := int((num + den - 1) / den) // ceil division
+		rowsPerValidator[i] = max(rows, minRows)
+	}
 
 	var seed [32]byte
 	copy(seed[:], commitment[:])
 
 	// chacha8 RNG with seed being the commitment
 	rng := rand.New(rand.NewChaCha8(seed))
 
-	// shuffle row indices with Fisher-Yates algorithm
+	// shuffle all totalRows indices with Fisher-Yates algorithm
 	// NOTE: std library Shuffle implements Fisher-Yates algorithm
 	rowsIndicies := make([]int, totalRows)
 	for i := range totalRows {
@@ -61,11 +79,17 @@ func (s Set) Assign(commitment rsema1d.Commitment, rowsPerValidator int) ShardMa
 		rowsIndicies[i], rowsIndicies[j] = rowsIndicies[j], rowsIndicies[i]
 	})
 
-	// assign rows to validators in a ShardMap
+	// assign rows to validators, wrapping around with modulo if total assigned exceeds totalRows
 	shardMap := make(ShardMap)
+	offset := 0
 	for i, validator := range s.Validators {
-		offset := i * rowsPerValidator
-		shardMap[validator] = rowsIndicies[offset : offset+rowsPerValidator]
+		rows := make([]int, rowsPerValidator[i])
+		for j := range rows {
+			// modulo ensures wrap-around when minRows causes over-assignment
+			rows[j] = rowsIndicies[(offset+j)%totalRows]
+		}
+		shardMap[validator] = rows
+		offset += rowsPerValidator[i]
 	}
 
 	return shardMap