Improve Tx Propagation

cmd/thor/node/packer_loop.go

@@ -16,6 +16,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/vechain/thor/packer"
 	"github.com/vechain/thor/thor"
+	"github.com/vechain/thor/tx"
 )
 
 func (n *Node) packerLoop(ctx context.Context) {
@@ -83,10 +84,10 @@ func (n *Node) packerLoop(ctx context.Context) {
 
 func (n *Node) pack(flow *packer.Flow) error {
 	txs := n.txPool.Executables()
-	var txsToRemove []thor.Bytes32
+	var txsToRemove []*tx.Transaction
 	defer func() {
-		for _, id := range txsToRemove {
-			n.txPool.Remove(id)
+		for _, tx := range txsToRemove {
+			n.txPool.Remove(tx.Hash(), tx.ID())
 		}
 	}()
 
@@ -99,7 +100,7 @@ func (n *Node) pack(flow *packer.Flow) error {
 			if packer.IsTxNotAdoptableNow(err) {
 				continue
 			}
-			txsToRemove = append(txsToRemove, tx.ID())
+			txsToRemove = append(txsToRemove, tx)
 		}
 	}
 

cmd/thor/node/tx_stash.go

@@ -6,6 +6,7 @@
 package node
 
 import (
+	"bytes"
 	"container/list"
 
 	"github.com/ethereum/go-ethereum/rlp"
@@ -27,7 +28,7 @@ func newTxStash(kv kv.GetPutter, maxSize int) *txStash {
 }
 
 func (ts *txStash) Save(tx *tx.Transaction) error {
-	has, err := ts.kv.Has(tx.ID().Bytes())
+	has, err := ts.kv.Has(tx.Hash().Bytes())
 	if err != nil {
 		return err
 	}
@@ -40,10 +41,10 @@ func (ts *txStash) Save(tx *tx.Transaction) error {
 		return err
 	}
 
-	if err := ts.kv.Put(tx.ID().Bytes(), data); err != nil {
+	if err := ts.kv.Put(tx.Hash().Bytes(), data); err != nil {
 		return err
 	}
-	ts.fifo.PushBack(tx.ID())
+	ts.fifo.PushBack(tx.Hash())
 	for ts.fifo.Len() > ts.maxSize {
 		keyToDelete := ts.fifo.Remove(ts.fifo.Front()).(thor.Bytes32).Bytes()
 		if err := ts.kv.Delete(keyToDelete); err != nil {
@@ -54,6 +55,7 @@ func (ts *txStash) Save(tx *tx.Transaction) error {
 }
 
 func (ts *txStash) LoadAll() tx.Transactions {
+	batch := ts.kv.NewBatch()
 	var txs tx.Transactions
 	iter := ts.kv.NewIterator(*kv.NewRangeWithBytesPrefix(nil))
 	for iter.Next() {
@@ -65,8 +67,19 @@ func (ts *txStash) LoadAll() tx.Transactions {
 			}
 		} else {
 			txs = append(txs, &tx)
-			ts.fifo.PushBack(tx.ID())
+			ts.fifo.PushBack(tx.Hash())
+
+			// Keys were tx ids.
+			// Here to remap values using tx hashes.
+			if !bytes.Equal(iter.Key(), tx.Hash().Bytes()) {
+				batch.Delete(iter.Key())
+				batch.Put(tx.Hash().Bytes(), iter.Value())
+			}
 		}
 	}
+
+	if err := batch.Write(); err != nil {
+		log.Warn("remap stashed txs", "err", err)
+	}
 	return txs
 }

cmd/thor/solo/solo.go

@@ -22,7 +22,6 @@ import (
 	"github.com/vechain/thor/logdb"
 	"github.com/vechain/thor/packer"
 	"github.com/vechain/thor/state"
-	"github.com/vechain/thor/thor"
 	"github.com/vechain/thor/tx"
 	"github.com/vechain/thor/txpool"
 )
@@ -118,10 +117,10 @@ func (s *Solo) loop(ctx context.Context) {
 
 func (s *Solo) packing(pendingTxs tx.Transactions) error {
 	best := s.chain.BestBlock()
-	var txsToRemove []thor.Bytes32
+	var txsToRemove []*tx.Transaction
 	defer func() {
-		for _, id := range txsToRemove {
-			s.txPool.Remove(id)
+		for _, tx := range txsToRemove {
+			s.txPool.Remove(tx.Hash(), tx.ID())
 		}
 	}()
 
@@ -142,7 +141,7 @@ func (s *Solo) packing(pendingTxs tx.Transactions) error {
 		case packer.IsTxNotAdoptableNow(err):
 			continue
 		default:
-			txsToRemove = append(txsToRemove, tx.ID())
+			txsToRemove = append(txsToRemove, tx)
 		}
 	}
 

comm/handle_rpc.go

@@ -69,8 +69,8 @@ func (c *Communicator) handleRPC(peer *Peer, msg *p2p.Msg, write func(interface{
 		if err := msg.Decode(&newTx); err != nil {
 			return errors.WithMessage(err, "decode msg")
 		}
-		peer.MarkTransaction(newTx.ID())
-		c.txPool.StrictlyAdd(newTx)
+		peer.MarkTransaction(newTx.Hash())
+		c.txPool.Add(newTx)
 		write(&struct{}{})
 	case proto.MsgGetBlockByID:
 		var blockID thor.Bytes32
@@ -146,10 +146,10 @@ func (c *Communicator) handleRPC(peer *Peer, msg *p2p.Msg, write func(interface{
 
 			for _, tx := range txsToSync.txs {
 				n++
-				if peer.IsTransactionKnown(tx.ID()) {
+				if peer.IsTransactionKnown(tx.Hash()) {
 					continue
 				}
-				peer.MarkTransaction(tx.ID())
+				peer.MarkTransaction(tx.Hash())
 				toSend = append(toSend, tx)
 				size += tx.Size()
 				if size >= maxTxSyncSize {

comm/peer.go

@@ -20,8 +20,9 @@ import (
 )
 
 const (
-	maxKnownTxs    = 32768 // Maximum transactions IDs to keep in the known list (prevent DOS)
-	maxKnownBlocks = 1024  // Maximum block IDs to keep in the known list (prevent DOS)
+	maxKnownTxs           = 32768 // Maximum transactions IDs to keep in the known list (prevent DOS)
+	maxKnownBlocks        = 1024  // Maximum block IDs to keep in the known list (prevent DOS)
+	knownTxMarkExpiration = 10    // Time in seconds to expire known tx mark
 )
 
 func init() {
@@ -82,8 +83,8 @@ func (p *Peer) UpdateHead(id thor.Bytes32, totalScore uint64) {
 }
 
 // MarkTransaction marks a transaction to known.
-func (p *Peer) MarkTransaction(id thor.Bytes32) {
-	p.knownTxs.Add(id, struct{}{})
+func (p *Peer) MarkTransaction(hash thor.Bytes32) {
+	p.knownTxs.Add(hash, time.Now().Unix())
 }
 
 // MarkBlock marks a block to known.
@@ -92,8 +93,12 @@ func (p *Peer) MarkBlock(id thor.Bytes32) {
 }
 
 // IsTransactionKnown returns if the transaction is known.
-func (p *Peer) IsTransactionKnown(id thor.Bytes32) bool {
-	return p.knownTxs.Contains(id)
+func (p *Peer) IsTransactionKnown(hash thor.Bytes32) bool {
+	ts, ok := p.knownTxs.Get(hash)
+	if !ok {
+		return false
+	}
+	return ts.(int64)+knownTxMarkExpiration > time.Now().Unix()
 }
 
 // IsBlockKnown returns if the block is known.

comm/sync.go

@@ -193,7 +193,7 @@ func (c *Communicator) syncTxs(peer *Peer) {
 		}
 
 		for _, tx := range result {
-			peer.MarkTransaction(tx.ID())
+			peer.MarkTransaction(tx.Hash())
 			c.txPool.StrictlyAdd(tx)
 			select {
 			case <-c.ctx.Done():

comm/txs_loop.go

@@ -24,12 +24,12 @@ func (c *Communicator) txsLoop() {
 			if txEv.Executable != nil && *txEv.Executable {
 				tx := txEv.Tx
 				peers := c.peerSet.Slice().Filter(func(p *Peer) bool {
-					return !p.IsTransactionKnown(tx.ID())
+					return !p.IsTransactionKnown(tx.Hash())
 				})
 
 				for _, peer := range peers {
 					peer := peer
-					peer.MarkTransaction(tx.ID())
+					peer.MarkTransaction(tx.Hash())
 					c.goes.Go(func() {
 						if err := proto.NotifyNewTx(c.ctx, peer, tx); err != nil {
 							peer.logger.Debug("failed to broadcast tx", "err", err)

tx/transaction.go

@@ -37,6 +37,7 @@ type Transaction struct {
 		unprovedWork atomic.Value
 		size         atomic.Value
 		intrinsicGas atomic.Value
+		hash         atomic.Value
 	}
 }
 
@@ -102,6 +103,19 @@ func (t *Transaction) ID() (id thor.Bytes32) {
 	return
 }
 
+// Hash returns hash of tx.
+// Unlike ID, it's the hash of RLP encoded tx.
+func (t *Transaction) Hash() (hash thor.Bytes32) {
+	if cached := t.cache.hash.Load(); cached != nil {
+		return cached.(thor.Bytes32)
+	}
+	defer func() { t.cache.hash.Store(hash) }()
+	hw := thor.NewBlake2b()
+	rlp.Encode(hw, t)
+	hw.Sum(hash[:0])
+	return
+}
+
 // UnprovedWork returns unproved work of this tx.
 // It returns 0, if tx is not signed.
 func (t *Transaction) UnprovedWork() (w *big.Int) {

txpool/tx_object_map.go

@@ -13,7 +13,7 @@ import (
 	"github.com/vechain/thor/tx"
 )
 
-// txObjectMap to maintain mapping of ID to tx object, and account quota.
+// txObjectMap to maintain mapping of tx hash to tx object, and account quota.
 type txObjectMap struct {
 	lock     sync.RWMutex
 	txObjMap map[thor.Bytes32]*txObject
@@ -27,18 +27,18 @@ func newTxObjectMap() *txObjectMap {
 	}
 }
 
-func (m *txObjectMap) Contains(txID thor.Bytes32) bool {
+func (m *txObjectMap) Contains(txHash thor.Bytes32) bool {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
-	_, found := m.txObjMap[txID]
+	_, found := m.txObjMap[txHash]
 	return found
 }
 
 func (m *txObjectMap) Add(txObj *txObject, limitPerAccount int) error {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 
-	if _, found := m.txObjMap[txObj.ID()]; found {
+	if _, found := m.txObjMap[txObj.Hash()]; found {
 		return nil
 	}
 
@@ -47,21 +47,21 @@ func (m *txObjectMap) Add(txObj *txObject, limitPerAccount int) error {
 	}
 
 	m.quota[txObj.Origin()]++
-	m.txObjMap[txObj.ID()] = txObj
+	m.txObjMap[txObj.Hash()] = txObj
 	return nil
 }
 
-func (m *txObjectMap) Remove(txID thor.Bytes32) bool {
+func (m *txObjectMap) Remove(txHash thor.Bytes32) bool {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 
-	if txObj, ok := m.txObjMap[txID]; ok {
+	if txObj, ok := m.txObjMap[txHash]; ok {
 		if m.quota[txObj.Origin()] > 1 {
 			m.quota[txObj.Origin()]--
 		} else {
 			delete(m.quota, txObj.Origin())
 		}
-		delete(m.txObjMap, txID)
+		delete(m.txObjMap, txHash)
 		return true
 	}
 	return false
@@ -93,13 +93,13 @@ func (m *txObjectMap) Fill(txObjs []*txObject) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	for _, txObj := range txObjs {
-		if _, found := m.txObjMap[txObj.ID()]; found {
+		if _, found := m.txObjMap[txObj.Hash()]; found {
 			continue
 		}
 		// skip account limit check
 
 		m.quota[txObj.Origin()]++
-		m.txObjMap[txObj.ID()] = txObj
+		m.txObjMap[txObj.Hash()] = txObj
 	}
 }
 

txpool/tx_object_map_test.go

@@ -41,13 +41,13 @@ func TestTxObjMap(t *testing.T) {
 	assert.Nil(t, m.Add(txObj3, 1))
 	assert.Equal(t, 2, m.Len())
 
-	assert.True(t, m.Contains(tx1.ID()))
-	assert.False(t, m.Contains(tx2.ID()))
-	assert.True(t, m.Contains(tx3.ID()))
+	assert.True(t, m.Contains(tx1.Hash()))
+	assert.False(t, m.Contains(tx2.Hash()))
+	assert.True(t, m.Contains(tx3.Hash()))
 
-	assert.True(t, m.Remove(tx1.ID()))
-	assert.False(t, m.Contains(tx1.ID()))
-	assert.False(t, m.Remove(tx2.ID()))
+	assert.True(t, m.Remove(tx1.Hash()))
+	assert.False(t, m.Contains(tx1.Hash()))
+	assert.False(t, m.Remove(tx2.Hash()))
 
 	assert.Equal(t, []*txObject{txObj3}, m.ToTxObjects())
 	assert.Equal(t, tx.Transactions{tx3}, m.ToTxs())