[#365] settlement/basic: Implement asset distribution

Signed-off-by: Alex Vanin <alexey@nspcc.ru>

pkg/innerring/processors/settlement/basic/collect.go

@@ -56,6 +56,11 @@ func (inc *IncomeSettlementContext) Collect() {
 		avg := inc.avgEstimation(cnrEstimations[i]) // average container size per node
 		total := calculateBasicSum(avg, cachedRate, len(cnrNodes))
 
+		// fill distribute asset table
+		for i := range cnrNodes {
+			inc.distributeTable.Put(cnrNodes[i].PublicKey(), avg)
+		}
+
 		inc.txTable.Transfer(&common.TransferTx{
 			From:   owner.Owner(),
 			To:     inc.bankOwner,

pkg/innerring/processors/settlement/basic/context.go

@@ -39,9 +39,13 @@ type (
 		container   common.ContainerStorage
 		placement   common.PlacementCalculator
 		exchange    common.Exchanger
+		accounts    common.AccountStorage
 
 		txTable   *common.TransferTable
 		bankOwner *owner.ID
+
+		// this table is not thread safe, make sure you use it with mu.Lock()
+		distributeTable *NodeSizeTable
 	}
 
 	IncomeSettlementContextPrms struct {
@@ -53,6 +57,7 @@ type (
 		Container   common.ContainerStorage
 		Placement   common.PlacementCalculator
 		Exchange    common.Exchanger
+		Accounts    common.AccountStorage
 	}
 )
 
@@ -63,16 +68,18 @@ func NewIncomeSettlementContext(p *IncomeSettlementContextPrms) (*IncomeSettleme
 	}
 
 	return &IncomeSettlementContext{
-		log:         p.Log,
-		epoch:       p.Epoch,
-		rate:        p.Rate,
-		estimations: p.Estimations,
-		balances:    p.Balances,
-		container:   p.Container,
-		placement:   p.Placement,
-		exchange:    p.Exchange,
-		txTable:     common.NewTransferTable(),
-		bankOwner:   bankingAccount,
+		log:             p.Log,
+		epoch:           p.Epoch,
+		rate:            p.Rate,
+		estimations:     p.Estimations,
+		balances:        p.Balances,
+		container:       p.Container,
+		placement:       p.Placement,
+		exchange:        p.Exchange,
+		accounts:        p.Accounts,
+		txTable:         common.NewTransferTable(),
+		bankOwner:       bankingAccount,
+		distributeTable: NewNodeSizeTable(),
 	}, nil
 }
 

pkg/innerring/processors/settlement/basic/distribute.go

@@ -1,6 +1,10 @@
 package basic
 
 import (
+	"encoding/hex"
+	"math/big"
+
+	"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/settlement/common"
 	"go.uber.org/zap"
 )
 
@@ -8,6 +12,8 @@ func (inc *IncomeSettlementContext) Distribute() {
 	inc.mu.Lock()
 	defer inc.mu.Unlock()
 
+	txTable := common.NewTransferTable()
+
 	bankBalance, err := inc.balances.Balance(inc.bankOwner)
 	if err != nil {
 		inc.log.Error("can't fetch balance of banking account",
@@ -16,5 +22,33 @@ func (inc *IncomeSettlementContext) Distribute() {
 		return
 	}
 
-	_ = bankBalance
+	total := inc.distributeTable.Total()
+
+	inc.distributeTable.Iterate(func(key []byte, n *big.Int) {
+		nodeOwner, err := inc.accounts.ResolveKey(nodeInfoWrapper(key))
+		if err != nil {
+			inc.log.Warn("can't transform public key to owner id",
+				zap.String("public_key", hex.EncodeToString(key)),
+				zap.String("error", err.Error()))
+
+			return
+		}
+
+		txTable.Transfer(&common.TransferTx{
+			From:   inc.bankOwner,
+			To:     nodeOwner,
+			Amount: normalizedValue(n, total, bankBalance),
+		})
+	})
+
+	common.TransferAssets(inc.exchange, txTable)
+}
+
+func normalizedValue(n, total, limit *big.Int) *big.Int {
+	if limit.Cmp(bigZero) == 0 {
+		return new(big.Int)
+	}
+
+	n.Mul(n, limit)
+	return n.Div(n, total)
 }

pkg/innerring/processors/settlement/basic/distribute_test.go

@@ -0,0 +1,54 @@
+package basic
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type normalizedValueCase struct {
+	name            string
+	n, total, limit uint64
+	expected        uint64
+}
+
+func TestNormalizedValues(t *testing.T) {
+	testCases := []normalizedValueCase{
+		{
+			name:     "zero limit",
+			n:        50,
+			total:    100,
+			limit:    0,
+			expected: 0,
+		},
+		{
+			name:     "scale down",
+			n:        50,
+			total:    100,
+			limit:    10,
+			expected: 5,
+		},
+		{
+			name:     "scale up",
+			n:        50,
+			total:    100,
+			limit:    1000,
+			expected: 500,
+		},
+	}
+
+	for _, testCase := range testCases {
+		testNormalizedValues(t, testCase)
+	}
+}
+
+func testNormalizedValues(t *testing.T, c normalizedValueCase) {
+	n := new(big.Int).SetUint64(c.n)
+	total := new(big.Int).SetUint64(c.total)
+	limit := new(big.Int).SetUint64(c.limit)
+	exp := new(big.Int).SetUint64(c.expected)
+
+	got := normalizedValue(n, total, limit)
+	require.Zero(t, exp.Cmp(got), c.name)
+}

pkg/innerring/processors/settlement/basic/util.go

@@ -0,0 +1,44 @@
+package basic
+
+import (
+	"math/big"
+)
+
+// NodeSizeTable is not thread safe, make sure it is accessed with external
+// locks or in single routine.
+type NodeSizeTable struct {
+	prices map[string]uint64
+	total  uint64
+}
+
+func (t *NodeSizeTable) Put(id []byte, avg uint64) {
+	t.prices[string(id)] += avg
+	t.total += avg
+}
+
+func (t *NodeSizeTable) Total() *big.Int {
+	return new(big.Int).SetUint64(t.total)
+}
+
+func (t *NodeSizeTable) Iterate(f func([]byte, *big.Int)) {
+	for k, v := range t.prices {
+		n := new(big.Int).SetUint64(v)
+		f([]byte(k), n)
+	}
+}
+
+func NewNodeSizeTable() *NodeSizeTable {
+	return &NodeSizeTable{
+		prices: make(map[string]uint64),
+	}
+}
+
+type nodeInfoWrapper []byte
+
+func (nodeInfoWrapper) Price() *big.Int {
+	panic("should not be used")
+}
+
+func (n nodeInfoWrapper) PublicKey() []byte {
+	return n
+}