Fix races in Merkle Tree (#12518)

cl/cltypes/solid/validator_set.go

@@ -99,14 +99,14 @@ func (v *ValidatorSet) Append(val Validator) {
 		v.expandBuffer(v.l + 1)
 		v.phase0Data = append(v.phase0Data, Phase0Data{})
 	}
+	if v.MerkleTree != nil {
+		v.MerkleTree.AppendLeaf()
+	}
 	v.zeroTreeHash(v.l)
 	copy(v.buffer[offset:], val)
 	v.phase0Data[v.l] = Phase0Data{} // initialize to empty.
 	v.attesterBits = append(v.attesterBits, 0x0)
 	v.l++
-	if v.MerkleTree != nil {
-		v.MerkleTree.AppendLeaf()
-	}
 }
 
 func (v *ValidatorSet) Cap() int {

cl/merkle_tree/merkle_tree.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"encoding/binary"
 	"io"
+	"sync"
+	"sync/atomic"
 
 	libcommon "github.com/erigontech/erigon-lib/common"
 	"github.com/erigontech/erigon-lib/common/length"
@@ -28,6 +30,9 @@ type MerkleTree struct {
 
 	hashBuf [64]byte // buffer to store the input for hash(hash1, hash2)
 	limit   *uint64  // Optional limit for the number of leaves (this will enable limit-oriented hashing)
+
+	dirtyLeaves []atomic.Bool
+	mu          sync.RWMutex
 }
 
 var _ HashTreeEncodable = (*MerkleTree)(nil)
@@ -50,10 +55,17 @@ func (m *MerkleTree) Initialize(leavesCount, maxTreeCacheDepth int, computeLeaf
 		m.limit = new(uint64)
 		*m.limit = *limitOptional
 	}
+	m.dirtyLeaves = make([]atomic.Bool, leavesCount)
 }
 
-// MarkLeafAsDirty resets the leaf at the given index, so that it will be recomputed on the next call to ComputeRoot.
 func (m *MerkleTree) MarkLeafAsDirty(idx int) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	m.dirtyLeaves[idx].Store(true)
+}
+
+// MarkLeafAsDirty resets the leaf at the given index, so that it will be recomputed on the next call to ComputeRoot.
+func (m *MerkleTree) markLeafAsDirty(idx int) {
 	for i := 0; i < len(m.layers); i++ {
 		currDivisor := 1 << (i + 1) // i+1 because the first layer is not the leaf layer
 		layerSize := (m.leavesCount + (currDivisor - 1)) / currDivisor
@@ -75,6 +87,8 @@ func (m *MerkleTree) MarkLeafAsDirty(idx int) {
 }
 
 func (m *MerkleTree) AppendLeaf() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	/*
 		Step 1: Append a new dirty leaf
 		Step 2: Extend each layer with the new leaf when needed (1.5x extension)
@@ -83,6 +97,7 @@ func (m *MerkleTree) AppendLeaf() {
 		m.extendLayer(i)
 	}
 	m.leavesCount++
+	m.dirtyLeaves = append(m.dirtyLeaves, atomic.Bool{})
 }
 
 // extendLayer extends the layer with the given index by 1.5x, by marking the new leaf as dirty.
@@ -122,10 +137,18 @@ func (m *MerkleTree) extendLayer(layerIdx int) {
 
 // ComputeRoot computes the root of the Merkle tree.
 func (m *MerkleTree) ComputeRoot() libcommon.Hash {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	var root libcommon.Hash
 	if len(m.layers) == 0 {
 		return ZeroHashes[0]
 	}
+	for idx := range m.dirtyLeaves {
+		if m.dirtyLeaves[idx].Load() {
+			m.markLeafAsDirty(idx)
+			m.dirtyLeaves[idx].Store(false)
+		}
+	}
 
 	if m.leavesCount == 0 {
 		if m.limit == nil {
@@ -180,6 +203,10 @@ func (m *MerkleTree) ComputeRoot() libcommon.Hash {
 }
 
 func (m *MerkleTree) CopyInto(other *MerkleTree) {
+	other.mu.Lock()
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	defer other.mu.Unlock()
 	other.computeLeaf = m.computeLeaf
 	other.layers = make([][]byte, len(m.layers))
 	for i := 0; i < len(m.layers); i++ {
@@ -188,6 +215,11 @@ func (m *MerkleTree) CopyInto(other *MerkleTree) {
 	}
 	other.leavesCount = m.leavesCount
 	other.limit = m.limit
+	other.dirtyLeaves = make([]atomic.Bool, len(m.dirtyLeaves))
+
+	for i := 0; i < len(m.dirtyLeaves); i++ {
+		other.dirtyLeaves[i].Store(m.dirtyLeaves[i].Load())
+	}
 }
 
 func (m *MerkleTree) finishHashing(lastLayerIdx int, root []byte) {

cl/phase1/core/state/raw/setters.go

@@ -398,6 +398,8 @@ func (b *BeaconState) AddInactivityScore(score uint64) {
 }
 
 func (b *BeaconState) SetValidatorInactivityScore(index int, score uint64) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	if index >= b.inactivityScores.Length() {
 		return ErrInvalidValidatorIndex
 	}

cl/phase1/forkchoice/fork_graph/fork_graph_disk.go

@@ -288,7 +288,7 @@ func (f *forkGraphDisk) AddChainSegment(signedBlock *cltypes.SignedBeaconBlock,
 		prevInactivityScores = libcommon.Copy(newState.RawInactivityScores())
 	}
 	// Execute the state
-	if invalidBlockErr := transition.TransitionState(newState, signedBlock, blockRewardsCollector, fullValidation); invalidBlockErr != nil {
+	if invalidBlockErr := transition.TransitionState(newState, signedBlock, blockRewardsCollector, false); invalidBlockErr != nil {
 		// Add block to list of invalid blocks
 		log.Warn("Invalid beacon block", "slot", block.Slot, "blockRoot", blockRoot, "reason", invalidBlockErr)
 		f.badBlocks.Store(libcommon.Hash(blockRoot), struct{}{})

cl/transition/impl/eth2/operations.go

@@ -862,10 +862,11 @@ func (I *impl) ProcessSlots(s abstract.BeaconState, slot uint64) error {
 
 		if (sSlot+1)%beaconConfig.SlotsPerEpoch == 0 {
 			start := time.Now()
+
 			if err := statechange.ProcessEpoch(s); err != nil {
 				return err
 			}
-			log.Trace(
+			log.Info(
 				"Processed new epoch successfully",
 				"epoch",
 				state.Epoch(s),

cl/transition/impl/eth2/statechange/process_inactivity_scores.go

@@ -50,6 +50,7 @@ func ProcessInactivityScores(s abstract.BeaconState, eligibleValidatorsIndicies
 		if !state.InactivityLeaking(s) {
 			score -= min(s.BeaconConfig().InactivityScoreRecoveryRate, score)
 		}
+
 		if err := s.SetValidatorInactivityScore(int(validatorIndex), score); err != nil {
 			return err
 		}