feat(txpool): calculate consumption when committing a new block

node/node.go

@@ -59,13 +59,13 @@ func NewNode(genDoc *genesis.Genesis, conf *config.Config,
 		eventCh = nil
 	}
 
-	txPool := txpool.NewTxPool(conf.TxPool, messageCh)
-
 	str, err := store.NewStore(conf.Store)
 	if err != nil {
 		return nil, err
 	}
 
+	txPool := txpool.NewTxPool(conf.TxPool, messageCh, str)
+
 	st, err := state.LoadOrNewState(genDoc, valKeys, str, txPool, eventCh)
 	if err != nil {
 		return nil, err

state/state.go

@@ -325,7 +325,6 @@
 		// Only one subsidy transaction per blk
 		if txs[i].IsSubsidyTx() {
 			st.logger.Error("found duplicated subsidy transaction", "tx", txs[i])
-			st.txPool.RemoveTx(txs[i].ID())
 			txs.Remove(i)
 			i--
 
@@ -434,9 +433,9 @@
 
 	st.store.SaveBlock(blk, cert)
 
-	// Remove transactions from pool
-	for _, trx := range blk.Transactions() {
-		st.txPool.RemoveTx(trx.ID())
+	// Remove transactions from pool and update consumption
+	if err = st.txPool.HandleCommittedBlock(blk); err != nil {
+		return err
 	}
 
 	if err := st.store.WriteBatch(); err != nil {

txpool/config.go

@@ -5,8 +5,9 @@ import (
 )
 
 type Config struct {
-	MaxSize int        `toml:"max_size"`
-	Fee     *FeeConfig `toml:"fee"`
+	MaxSize           int        `toml:"max_size"`
+	ConsumptionBlocks uint32     `toml:"-"`
+	Fee               *FeeConfig `toml:"fee"`
 }
 
 type FeeConfig struct {
@@ -17,8 +18,9 @@ type FeeConfig struct {
 
 func DefaultConfig() *Config {
 	return &Config{
-		MaxSize: 1000,
-		Fee:     DefaultFeeConfig(),
+		MaxSize:           1000,
+		ConsumptionBlocks: 8640,
+		Fee:               DefaultFeeConfig(),
 	}
 }
 

txpool/interface.go

@@ -23,5 +23,5 @@
 	SetNewSandboxAndRecheck(sb sandbox.Sandbox)
 	AppendTxAndBroadcast(trx *tx.Tx) error
 	AppendTx(trx *tx.Tx) error
-	RemoveTx(id tx.ID)
+	HandleCommittedBlock(block *block.Block) error
 }

txpool/mock.go

@@ -79,6 +79,10 @@
 	}
 }
 
+func (m *MockTxPool) HandleCommittedBlock(_ *block.Block) error {
+	return nil
+}
+
 func (m *MockTxPool) PrepareBlockTransactions() block.Txs {
 	txs := make([]*tx.Tx, m.Size())
 	copy(txs, m.Txs)

txpool/txpool.go

@@ -4,6 +4,9 @@
 	"fmt"
 	"sync"
 
+	"github.com/pactus-project/pactus/crypto"
+	"github.com/pactus-project/pactus/store"
+
 	"github.com/pactus-project/pactus/execution"
 	"github.com/pactus-project/pactus/sandbox"
 	"github.com/pactus-project/pactus/sync/bundle/message"
@@ -19,14 +22,16 @@
 type txPool struct {
 	lk sync.RWMutex
 
-	config      *Config
-	sandbox     sandbox.Sandbox
-	pools       map[payload.Type]pool
-	broadcastCh chan message.Message
-	logger      *logger.SubLogger
+	config         *Config
+	sandbox        sandbox.Sandbox
+	pools          map[payload.Type]pool
+	consumptionMap map[crypto.Address]uint32
+	broadcastCh    chan message.Message
+	strReader      store.Reader
+	logger         *logger.SubLogger
 }
 
-func NewTxPool(conf *Config, broadcastCh chan message.Message) TxPool {
+func NewTxPool(conf *Config, broadcastCh chan message.Message, storeReader store.Reader) TxPool {
 	pools := make(map[payload.Type]pool)
 	pools[payload.TypeTransfer] = newPool(conf.transferPoolSize(), conf.minFee())
 	pools[payload.TypeBond] = newPool(conf.bondPoolSize(), conf.minFee())
@@ -35,9 +40,11 @@
 	pools[payload.TypeSortition] = newPool(conf.sortitionPoolSize(), 0)
 
 	pool := &txPool{
-		config:      conf,
-		pools:       pools,
-		broadcastCh: broadcastCh,
+		config:         conf,
+		pools:          pools,
+		consumptionMap: make(map[crypto.Address]uint32),
+		strReader:      storeReader,
+		broadcastCh:    broadcastCh,
 	}
 
 	pool.logger = logger.NewSubLogger("_pool", pool)
@@ -133,10 +140,57 @@
 	return nil
 }
 
-func (p *txPool) RemoveTx(id tx.ID) {
+func (p *txPool) HandleCommittedBlock(blk *block.Block) error {
 	p.lk.Lock()
 	defer p.lk.Unlock()
 
+	for _, trx := range blk.Transactions() {
+		p.removeTx(trx.ID())
+
+		p.handleIncreaseConsumption(trx)
+	}
+
+	return p.handleDecreaseConsumption(blk.Height())
+}
+
+func (p *txPool) handleIncreaseConsumption(trx *tx.Tx) {
+	if trx.IsTransferTx() || trx.IsBondTx() || trx.IsWithdrawTx() {
+		signer := trx.Payload().Signer()
+
+		// retrieve existing consumption or start with 0
+		p.consumptionMap[signer] = p.consumptionMap[signer] + uint32(trx.SerializeSize())
+	}
+}
+
+func (p *txPool) handleDecreaseConsumption(height uint32) error {
+	if height <= p.config.ConsumptionBlocks {
+		return nil
+	}
+
+	oldConsumptionHeight := height - p.config.ConsumptionBlocks
+	committedBlock, err := p.strReader.Block(oldConsumptionHeight)
+	if err != nil {
+		return err
+	}
+
+	blk, err := committedBlock.ToBlock()
+	if err != nil {
+		return err
+	}
+
+	for _, trx := range blk.Transactions() {
+		if trx.IsTransferTx() || trx.IsBondTx() || trx.IsWithdrawTx() {
+			signer := trx.Payload().Signer()
+			if v, ok := p.consumptionMap[signer]; ok {
+				p.consumptionMap[signer] = v - uint32(trx.SerializeSize())
+			}
+		}
+	}
+
+	return nil
+}
+
+func (p *txPool) removeTx(id tx.ID) {
 	for _, pool := range p.pools {
 		if pool.list.Remove(id) {
 			break

txpool/txpool_test.go

@@ -5,6 +5,8 @@
 	"testing"
 	"time"
 
+	"github.com/pactus-project/pactus/store"
+
 	"github.com/pactus-project/pactus/execution"
 	"github.com/pactus-project/pactus/sandbox"
 	"github.com/pactus-project/pactus/sync/bundle/message"
@@ -22,12 +24,14 @@
 
 	pool    *txPool
 	sandbox *sandbox.MockSandbox
+	str     *store.MockStore
 	ch      chan message.Message
 }
 
 func testConfig() *Config {
 	return &Config{
-		MaxSize: 10,
+		MaxSize:           10,
+		ConsumptionBlocks: 3,
 		Fee: &FeeConfig{
 			FixedFee:   0.000001,
 			DailyLimit: 280,
@@ -44,7 +48,8 @@
 	ch := make(chan message.Message, 10)
 	sb := sandbox.MockingSandbox(ts)
 	config := testConfig()
-	p := NewTxPool(config, ch)
+	mockStore := store.MockingStore(ts)
+	p := NewTxPool(config, ch, mockStore)
 	p.SetNewSandboxAndRecheck(sb)
 	pool := p.(*txPool)
 	assert.NotNil(t, pool)
@@ -53,6 +58,7 @@
 		TestSuite: ts,
 		pool:      pool,
 		sandbox:   sb,
+		str:       mockStore,
 		ch:        ch,
 	}
 }
@@ -96,11 +102,36 @@
 	// Appending the same transaction again, should not return any error
 	assert.NoError(t, td.pool.AppendTx(testTrx))
 
-	td.pool.RemoveTx(testTrx.ID())
+	td.pool.removeTx(testTrx.ID())
 	assert.False(t, td.pool.HasTx(testTrx.ID()), "Transaction should be removed")
 	assert.Nil(t, td.pool.PendingTx(testTrx.ID()))
 }
 
+func TestCalculatingConsumption(t *testing.T) {
+	td := setup(t)
+
+	_, prv := td.TestSuite.RandEd25519KeyPair()
+	signer := prv
+
+	for i := uint32(1); i < 10; i++ {
+		txr := td.GenerateTestTransferTx(func(tm *testsuite.TransactionMaker) {
+			tm.Signer = signer
+		})
+
+		blk, cert := td.TestSuite.GenerateTestBlock(i, func(bm *testsuite.BlockMaker) {
+			bm.Txs = []*tx.Tx{txr}
+		})
+
+		td.str.SaveBlock(blk, cert)
+
+		err := td.pool.HandleCommittedBlock(blk)
+		require.NoError(t, err)
+
+		//t.Log(td.pool.consumptionMap[txr.Payload().Signer()])
+	}
+
+}
+
 func TestAppendInvalidTransaction(t *testing.T) {
 	td := setup(t)