[NOD-1524] Add lock to consensus (#1014)

domain/consensus/consensus.go

@@ -1,11 +1,14 @@
 package consensus
 
 import (
+	"sync"
+
 	"github.com/kaspanet/kaspad/domain/consensus/model"
 	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
 )
 
 type consensus struct {
+	lock            *sync.RWMutex
 	databaseContext model.DBReader
 
 	blockProcessor        model.BlockProcessor
@@ -27,18 +30,27 @@ type consensus struct {
 func (s *consensus) BuildBlock(coinbaseData *externalapi.DomainCoinbaseData,
 	transactions []*externalapi.DomainTransaction) (*externalapi.DomainBlock, error) {
 
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.blockBuilder.BuildBlock(coinbaseData, transactions)
 }
 
 // ValidateAndInsertBlock validates the given block and, if valid, applies it
 // to the current state
 func (s *consensus) ValidateAndInsertBlock(block *externalapi.DomainBlock) error {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	return s.blockProcessor.ValidateAndInsertBlock(block)
 }
 
 // ValidateTransactionAndPopulateWithConsensusData validates the given transaction
 // and populates it with any missing consensus data
 func (s *consensus) ValidateTransactionAndPopulateWithConsensusData(transaction *externalapi.DomainTransaction) error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	err := s.transactionValidator.ValidateTransactionInIsolation(transaction)
 	if err != nil {
 		return err
@@ -63,14 +75,23 @@ func (s *consensus) ValidateTransactionAndPopulateWithConsensusData(transaction
 }
 
 func (s *consensus) GetBlock(blockHash *externalapi.DomainHash) (*externalapi.DomainBlock, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.blockStore.Block(s.databaseContext, blockHash)
 }
 
 func (s *consensus) GetBlockHeader(blockHash *externalapi.DomainHash) (*externalapi.DomainBlockHeader, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.blockHeaderStore.BlockHeader(s.databaseContext, blockHash)
 }
 
 func (s *consensus) GetBlockInfo(blockHash *externalapi.DomainHash) (*externalapi.BlockInfo, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	blockInfo := &externalapi.BlockInfo{}
 
 	exists, err := s.blockStatusStore.Exists(s.databaseContext, blockHash)
@@ -98,22 +119,37 @@ func (s *consensus) GetBlockInfo(blockHash *externalapi.DomainHash) (*externalap
 }
 
 func (s *consensus) GetHashesBetween(lowHash, highHash *externalapi.DomainHash) ([]*externalapi.DomainHash, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.syncManager.GetHashesBetween(lowHash, highHash)
 }
 
 func (s *consensus) GetMissingBlockBodyHashes(highHash *externalapi.DomainHash) ([]*externalapi.DomainHash, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.syncManager.GetMissingBlockBodyHashes(highHash)
 }
 
 func (s *consensus) GetPruningPointUTXOSet() ([]byte, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.pruningStore.PruningPointSerializedUTXOSet(s.databaseContext)
 }
 
 func (s *consensus) SetPruningPointUTXOSet(serializedUTXOSet []byte) error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.consensusStateManager.SetPruningPointUTXOSet(serializedUTXOSet)
 }
 
 func (s *consensus) GetVirtualSelectedParent() (*externalapi.DomainBlock, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	virtualGHOSTDAGData, err := s.ghostdagDataStore.Get(s.databaseContext, model.VirtualBlockHash)
 	if err != nil {
 		return nil, err
@@ -122,13 +158,22 @@ func (s *consensus) GetVirtualSelectedParent() (*externalapi.DomainBlock, error)
 }
 
 func (s *consensus) CreateBlockLocator(lowHash, highHash *externalapi.DomainHash) (externalapi.BlockLocator, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.syncManager.CreateBlockLocator(lowHash, highHash)
 }
 
 func (s *consensus) FindNextBlockLocatorBoundaries(blockLocator externalapi.BlockLocator) (lowHash, highHash *externalapi.DomainHash, err error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.syncManager.FindNextBlockLocatorBoundaries(blockLocator)
 }
 
 func (s *consensus) GetSyncInfo() (*externalapi.SyncInfo, error) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.syncManager.GetSyncInfo()
 }

domain/consensus/factory.go

@@ -1,6 +1,8 @@
 package consensus
 
 import (
+	"sync"
+
 	consensusdatabase "github.com/kaspanet/kaspad/domain/consensus/database"
 	"github.com/kaspanet/kaspad/domain/consensus/datastructures/acceptancedatastore"
 	"github.com/kaspanet/kaspad/domain/consensus/datastructures/blockheaderstore"
@@ -243,6 +245,7 @@ func (f *factory) NewConsensus(dagParams *dagconfig.Params, db infrastructuredat
 		headerTipsStore)
 
 	c := &consensus{
+		lock:            &sync.RWMutex{},
 		databaseContext: dbManager,
 
 		blockProcessor:        blockProcessor,

domain/consensus/test_consensus.go

@@ -13,5 +13,9 @@ type testConsensus struct {
 func (tc *testConsensus) BuildBlockWithParents(parentHashes []*externalapi.DomainHash, coinbaseData *externalapi.DomainCoinbaseData,
 	transactions []*externalapi.DomainTransaction) (*externalapi.DomainBlock, error) {
 
+	// Require write lock because BuildBlockWithParents stages temporary data
+	tc.lock.Lock()
+	defer tc.lock.Unlock()
+
 	return tc.testBlockBuilder.BuildBlockWithParents(parentHashes, coinbaseData, transactions)
 }