Fix data race(s) in replication code and tests

This commit fixes two data races:

1. We send requests to nodes for blocks asynchronously but also process messages in parallel, without synchronization.

2. The 'collectNotarizationComm' communication mock can be invoked from the test without synchronizing with the communication goroutines.

Signed-off-by: Yacov Manevich <yacov.manevich@avalabs.org>

Author: yacovm

epoch.go

@@ -183,7 +183,7 @@ func (e *Epoch) init() error {
 	e.maxPendingBlocks = DefaultMaxPendingBlocks
 	e.eligibleNodeIDs = make(map[string]struct{}, len(e.nodes))
 	e.futureMessages = make(messagesFromNode, len(e.nodes))
-	e.replicationState = NewReplicationState(e.Logger, e.Comm, e.ID, e.maxRoundWindow, e.ReplicationEnabled, e.StartTime)
+	e.replicationState = NewReplicationState(e.Logger, e.Comm, e.ID, e.maxRoundWindow, e.ReplicationEnabled, e.StartTime, &e.lock)
 	e.timeoutHandler = NewTimeoutHandler(e.Logger, e.StartTime, e.nodes)
 
 	for _, node := range e.nodes {

replication.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"math"
 	"slices"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
@@ -40,6 +41,7 @@ func newSignedSequenceFromRound(round QuorumRound) (*signedSequence, error) {
 }
 
 type ReplicationState struct {
+	lock *sync.Mutex
 	logger         Logger
 	enabled        bool
 	maxRoundWindow uint64
@@ -61,8 +63,9 @@ type ReplicationState struct {
 	timeoutHandler *TimeoutHandler
 }
 
-func NewReplicationState(logger Logger, comm Communication, id NodeID, maxRoundWindow uint64, enabled bool, start time.Time) *ReplicationState {
+func NewReplicationState(logger Logger, comm Communication, id NodeID, maxRoundWindow uint64, enabled bool, start time.Time, lock *sync.Mutex) *ReplicationState {
 	return &ReplicationState{
+		lock:                 lock,
 		logger:               logger,
 		enabled:              enabled,
 		comm:                 comm,
@@ -157,6 +160,8 @@ func (r *ReplicationState) sendRequestToNode(start uint64, end uint64, nodes []N
 
 func (r *ReplicationState) createReplicationTimeoutTask(start, end uint64, nodes []NodeID, index int) *TimeoutTask {
 	taskFunc := func() {
+		r.lock.Lock()
+		defer r.lock.Unlock()
 		r.sendRequestToNode(start, end, nodes, (index+1)%len(nodes))
 	}
 	timeoutTask := &TimeoutTask{

replication_timeout_test.go

@@ -327,6 +327,9 @@ func (c *collectNotarizationComm) Broadcast(msg *simplex.Message) {
 }
 
 func (c *collectNotarizationComm) removeFinalizationsFromReplicationResponses(msg *simplex.Message, from, to simplex.NodeID) bool {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
 	if msg.VerifiedReplicationResponse != nil || msg.ReplicationResponse != nil {
 		newData := make([]simplex.VerifiedQuorumRound, 0, len(msg.VerifiedReplicationResponse.Data))