Resend Replication Requests After a Timeout

epoch.go

@@ -24,7 +24,8 @@ const (
 	DefaultMaxRoundWindow   = 10
 	DefaultMaxPendingBlocks = 20
 
-	DefaultMaxProposalWaitTime = 5 * time.Second
+	DefaultMaxProposalWaitTime       = 5 * time.Second
+	DefaultReplicationRequestTimeout = 5 * time.Second
 )
 
 type EmptyVoteSet struct {
@@ -105,6 +106,7 @@ func NewEpoch(conf EpochConfig) (*Epoch, error) {
 // AdvanceTime hints the engine that the given amount of time has passed.
 func (e *Epoch) AdvanceTime(t time.Time) {
 	e.monitor.AdvanceTime(t)
+	e.replicationState.AdvanceTime(t)
 }
 
 // HandleMessage notifies the engine about a reception of a message.
@@ -173,7 +175,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.replicationState = NewReplicationState(e.Logger, e.Comm, e.ID, e.maxRoundWindow, e.ReplicationEnabled, e.StartTime)
 
 	for _, node := range e.nodes {
 		e.futureMessages[string(node)] = make(map[uint64]*messagesForRound)
@@ -2351,6 +2353,8 @@ func (e *Epoch) handleReplicationResponse(resp *ReplicationResponse, from NodeID
 		return nil
 	}
 
+	e.replicationState.receivedReplicationResponse(resp.Data, from)
+
 	return e.processReplicationState()
 }
 

epoch_multinode_test.go

@@ -271,6 +271,7 @@ func (tw *testWAL) containsEmptyNotarization(round uint64) bool {
 
 // messageFilter defines a function that filters
 // certain messages from being sent or broadcasted.
+// a message filter should return true if the message is allowed to be sent
 type messageFilter func(*Message, NodeID) bool
 
 // allowAllMessages allows every message to be sent

monitor_test.go

@@ -4,12 +4,13 @@
 package simplex
 
 import (
-	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 func TestMonitorDoubleClose(t *testing.T) {

replication.go

@@ -6,6 +6,8 @@ package simplex
 import (
 	"fmt"
 	"math"
+	"slices"
+	"time"
 
 	"go.uber.org/zap"
 )
@@ -55,19 +57,26 @@ type ReplicationState struct {
 
 	// request iterator
 	requestIterator int
+
+	timeoutHandler *TimeoutHandler
 }
 
-func NewReplicationState(logger Logger, comm Communication, id NodeID, maxRoundWindow uint64, enabled bool) *ReplicationState {
+func NewReplicationState(logger Logger, comm Communication, id NodeID, maxRoundWindow uint64, enabled bool, start time.Time) *ReplicationState {
 	return &ReplicationState{
 		logger:               logger,
 		enabled:              enabled,
 		comm:                 comm,
 		id:                   id,
 		maxRoundWindow:       maxRoundWindow,
 		receivedQuorumRounds: make(map[uint64]QuorumRound),
+		timeoutHandler:       NewTimeoutHandler(logger, start, comm.ListNodes()),
 	}
 }
 
+func (r *ReplicationState) AdvanceTime(now time.Time) {
+	r.timeoutHandler.Tick(now)
+}
+
 // isReplicationComplete returns true if we have finished the replication process.
 // The process is considered finished once [currentRound] has caught up to the highest round received.
 func (r *ReplicationState) isReplicationComplete(nextSeqToCommit uint64, currentRound uint64) bool {
@@ -111,6 +120,7 @@ func (r *ReplicationState) sendReplicationRequests(start uint64, end uint64) {
 	numSeqs := end + 1 - start
 	seqsPerNode := numSeqs / uint64(numNodes)
 
+	r.logger.Debug("Distributing replication requests", zap.Uint64("start", start), zap.Uint64("end", end), zap.Stringer("nodes", NodeIDs(nodes)))
 	// Distribute sequences evenly among nodes in round-robin fashion
 	for i := range numNodes {
 		nodeIndex := (r.requestIterator + i) % numNodes
@@ -122,16 +132,20 @@ func (r *ReplicationState) sendReplicationRequests(start uint64, end uint64) {
 			nodeEnd = end
 		}
 
-		r.sendRequestToNode(nodeStart, nodeEnd, nodes[nodeIndex])
+		r.sendRequestToNode(nodeStart, nodeEnd, nodes, nodeIndex)
 	}
 
+	r.lastSequenceRequested = end
 	// next time we send requests, we start with a different permutation
 	r.requestIterator++
 }
 
-func (r *ReplicationState) sendRequestToNode(start uint64, end uint64, node NodeID) {
+// sendRequestToNode requests the sequences [start, end] from nodes[index].
+// In case the nodes[index] does not respond, we create a timeout that will
+// re-send the request.
+func (r *ReplicationState) sendRequestToNode(start uint64, end uint64, nodes []NodeID, index int) {
 	r.logger.Debug("Requesting missing finalization certificates ",
-		zap.Stringer("from", node),
+		zap.Stringer("from", nodes[index]),
 		zap.Uint64("start", start),
 		zap.Uint64("end", end))
 	seqs := make([]uint64, (end+1)-start)
@@ -144,8 +158,83 @@ func (r *ReplicationState) sendRequestToNode(start uint64, end uint64, node Node
 	}
 	msg := &Message{ReplicationRequest: request}
 
-	r.lastSequenceRequested = end
-	r.comm.SendMessage(msg, node)
+	task := r.createReplicationTimeoutTask(start, end, nodes, index)
+
+	r.timeoutHandler.AddTask(task)
+
+	r.comm.SendMessage(msg, nodes[index])
+}
+
+func (r *ReplicationState) createReplicationTimeoutTask(start, end uint64, nodes []NodeID, index int) *TimeoutTask {
+	taskFunc := func() {
+		r.sendRequestToNode(start, end, nodes, (index+1)%len(nodes))
+	}
+	timeoutTask := &TimeoutTask{
+		Start:    start,
+		End:      end,
+		NodeID:   nodes[index],
+		TaskID:   getTimeoutID(start, end),
+		Task:     taskFunc,
+		Deadline: r.timeoutHandler.GetTime().Add(DefaultReplicationRequestTimeout),
+	}
+
+	return timeoutTask
+}
+
+// receivedReplicationResponse notifies the task handler a response was received. If the response
+// was incomplete(meaning our timeout expected more seqs), then we will create a new timeout
+// for the missing sequences and send the request to a different node.
+func (r *ReplicationState) receivedReplicationResponse(data []QuorumRound, node NodeID) {
+	seqs := make([]uint64, 0, len(data))
+
+	for _, qr := range data {
+		seqs = append(seqs, qr.GetSequence())
+	}
+
+	slices.Sort(seqs)
+
+	task := r.timeoutHandler.FindTask(node, seqs)
+	if task == nil {
+		r.logger.Debug("Could not find a timeout task associated with the replication response", zap.Stringer("from", node))
+		return
+	}
+	r.timeoutHandler.RemoveTask(node, task.TaskID)
+
+	// we found the timeout, now make sure all seqs were returned
+	missing := findMissingNumbersInRange(task.Start, task.End, seqs)
+	if len(missing) == 0 {
+		return
+	}
+
+	// if not all sequences were returned, create new timeouts
+	r.logger.Debug("Received missing sequences in the replication response", zap.Stringer("from", node), zap.Any("missing", missing))
+	nodes := r.highestSequenceObserved.signers.Remove(r.id)
+	numNodes := len(nodes)
+	segments := CompressSequences(missing)
+	for i, seqs := range segments {
+		index := i % numNodes
+		newTask := r.createReplicationTimeoutTask(seqs.Start, seqs.End, nodes, index)
+		r.timeoutHandler.AddTask(newTask)
+	}
+}
+
+// findMissingNumbersInRange finds numbers in an array constructed by [start...end] that are not in [nums]
+// ex. (3, 10, [1,2,3,4,5,6]) -> [7,8,9,10]
+func findMissingNumbersInRange(start, end uint64, nums []uint64) []uint64 {
+	numMap := make(map[uint64]struct{})
+	for _, num := range nums {
+		numMap[num] = struct{}{}
+	}
+
+	var result []uint64
+
+	for i := start; i <= end; i++ {
+		if _, exists := numMap[i]; !exists {
+			result = append(result, i)
+		}
+	}
+
+	return result
 }
 
 func (r *ReplicationState) replicateBlocks(fCert *FinalizationCertificate, nextSeqToCommit uint64) {
@@ -197,6 +286,8 @@ func (r *ReplicationState) StoreQuorumRound(round QuorumRound) {
 
 		r.highestSequenceObserved = signedSeq
 	}
+
+	r.logger.Debug("Stored quorum round ", zap.Stringer("qr", &round))
 	r.receivedQuorumRounds[round.GetRound()] = round
 }
 

replication_test.go

@@ -41,7 +41,7 @@ func testReplication(t *testing.T, startSeq uint64, nodes []simplex.NodeID) {
 	bb := newTestControlledBlockBuilder(t)
 	net := newInMemNetwork(t, nodes)
 
-	// initiate a network with 4 nodes. one node is behind by 8 blocks
+	// initiate a network with 4 nodes. one node is behind by startSeq blocks
 	storageData := createBlocks(t, nodes, &bb.testBlockBuilder, startSeq)
 	testEpochConfig := &testNodeConfig{
 		initialStorage:     storageData,