fix replication requests exceeding maxRoundWindow + splitting resends

.gitignore

@@ -32,4 +32,4 @@ go.work.sum
 .idea/
 
 # tmp folder
-/tmp
+/tmp

simplex/epoch.go

@@ -30,6 +30,7 @@ const (
 	DefaultEmptyVoteRebroadcastTimeout    = 5 * time.Second
 	DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second
 	EmptyVoteTimeoutID                    = "rebroadcast_empty_vote"
+	maxItemCountPerRequest                = 10 // max number of rounds or sequences that fit in one replication request
 )
 
 type EmptyVoteSet struct {

simplex/replication_state.go

@@ -235,7 +235,7 @@ func (r *ReplicationState) ResendFinalizationRequest(seq uint64, signers []commo
 	// because we are resending because the block failed to verify, we should remove the stored quorum round
 	// so that we can try to get a new block & finalization
 	r.DeleteSeq(seq)
-	r.finalizationRequestor.sendRequestToNode(seq, seq, signers[index])
+	r.finalizationRequestor.sendRequestToNode([]uint64{seq}, signers[index])
 }
 
 // CreateDependencyTasks creates tasks to refetch the given parent digest and empty rounds. If there are no

simplex/replication_timeout_test.go

@@ -677,3 +677,54 @@ func TestReplicationResendsFinalizedBlocksThatFailedVerification(t *testing.T) {
 	require.Equal(t, uint64(1), storage.NumBlocks())
 	require.Equal(t, block, storedBlock)
 }
+
+func TestReplicationResendSplitsRequests(t *testing.T) {
+	// ensures that when replication requests time out, the missing sequences
+	// are re-requested split across the nodes that signed the highest observed quorum, just like the initial requests are.
+	nodes := []common.NodeID{{1}, {2}, {3}, {4}}
+	startSeq := 2 * uint64(simplex.DefaultMaxRoundWindow)
+
+	net := testutil.NewControlledNetwork(t, nodes)
+	storageData := createBlocks(t, nodes, startSeq)
+
+	newNodeConfig := func(from common.NodeID) *testutil.TestNodeConfig {
+		comm := testutil.NewTestComm(from, net.BasicInMemoryNetwork, rejectReplicationRequests)
+		return &testutil.TestNodeConfig{
+			InitialStorage:     storageData,
+			Comm:               comm,
+			ReplicationEnabled: true,
+		}
+	}
+
+	testutil.NewControlledSimplexNode(t, nodes[0], net, newNodeConfig(nodes[0]))
+	testutil.NewControlledSimplexNode(t, nodes[1], net, newNodeConfig(nodes[1]))
+	testutil.NewControlledSimplexNode(t, nodes[2], net, newNodeConfig(nodes[2]))
+	laggingNode := testutil.NewControlledSimplexNode(t, nodes[3], net, &testutil.TestNodeConfig{
+		ReplicationEnabled: true,
+		// twice the responder's window: the timed-out sequences span more
+		// than a responder is willing to server in a single request
+		MaxRoundWindow: startSeq,
+	})
+
+	net.StartInstances()
+	defer net.StopInstances()
+	net.TriggerLeaderBlockBuilder(startSeq)
+
+	// the healthy nodes commit, the lagging node's replciation responses
+	// were all dropped, so it is stuck at 0.
+	for _, n := range net.Instances {
+		if n.E.ID.Equals(laggingNode.E.ID) {
+			continue
+		}
+		n.Storage.WaitForBlockCommit(startSeq)
+	}
+	require.Equal(t, uint64(0), laggingNode.Storage.NumBlocks())
+	// wait to register the timeout before healing
+	time.Sleep(100 * time.Millisecond)
+	// network healed, responsed are no longer dropped
+	net.SetAllNodesMessageFilter(testutil.AllowAllMessages)
+	// trigger the resend of all timed-out responses. If the resend path doesn't split/ doesn't account for maxRoundWindow
+	// this test would hang forever because the responses would be dropped silently.
+	laggingNode.E.AdvanceTime(laggingNode.E.StartTime.Add(simplex.DefaultReplicationRequestTimeout * 2))
+	laggingNode.Storage.WaitForBlockCommit(startSeq)
+}

simplex/requestor.go

@@ -5,6 +5,7 @@ package simplex
 
 import (
 	"math"
+	"slices"
 	"sync"
 	"time"
 
@@ -122,17 +123,16 @@ func (r *requestor) advanceTime(now time.Time) {
 	r.timeoutHandler.Tick(now)
 }
 
+// resendReplicationRequests re-sends requests for [missingIds], the sequences
+// or rounds that were previously requested but not received before their
+// timeout expired.
 func (r *requestor) resendReplicationRequests(missingIds []uint64) {
 	// we call this function in the timeout handler goroutine, so we need to
 	// ensure we don't have concurrent access to highestObserved
 	r.epochLock.Lock()
 	defer r.epochLock.Unlock()
 
-	segments := CompressSequences(missingIds)
-
-	r.sendSegments(segments)
-
-	r.requestIterator++
+	r.sendRequests(missingIds)
 }
 
 // observedSignedQuorum is called when we observe a signed quorum for a future round/sequence.
@@ -157,76 +157,78 @@ func (r *requestor) observedSignedQuorum(observed *signedQuorum, currentSeqOrRou
 // maybeSendMoreReplicationRequests checks if we need to send more replication requests given an observed quorum.
 // it limits the amount of outstanding requests to be at most [maxRoundWindow] ahead of [currentSeqOrRound].
 func (r *requestor) sendMoreReplicationRequests(observedSeqOrRound, currentSeqOrRound uint64) {
-	start := math.Max(float64(currentSeqOrRound), float64(r.highestRequested))
+	start := uint64(math.Max(float64(currentSeqOrRound), float64(r.highestRequested)))
 	// we limit the number of outstanding requests to be at most maxRoundWindow ahead of nextSeqToCommit
-	end := math.Min(float64(observedSeqOrRound), float64(r.maxRoundWindow+currentSeqOrRound-1))
+	end := uint64(math.Min(float64(observedSeqOrRound), float64(r.maxRoundWindow+currentSeqOrRound-1)))
 
+	if start > end {
+		return
+	}
+	seqsOrRounds := make([]uint64, 0, end-start+1)
+	for i := start; i <= end; i++ {
+		seqsOrRounds = append(seqsOrRounds, i)
+	}
 	r.logger.Debug("Node is behind, attempting to request missing values", zap.Uint64("value", observedSeqOrRound), zap.Uint64("start", uint64(start)), zap.Uint64("end", uint64(end)), zap.Bool("seq requestor", r.replicateSeqs))
-	r.sendReplicationRequests(uint64(start), uint64(end))
-}
-
-// sendReplicationRequests sends requests for missing sequences for the
-// range of sequences [start, end] <- inclusive. It does so by splitting the
-// range of sequences equally amount the nodes that have signed [highestObserved].
-func (r *requestor) sendReplicationRequests(start uint64, end uint64) {
-	nodes := r.highestObserved.signers
-	numNodes := len(nodes)
-
-	seqRequests := DistributeSequenceRequests(start, end, numNodes)
-	r.logger.Debug("Distributing replication requests", zap.Uint64("start", start), zap.Uint64("end", end), zap.Stringer("nodes", common.NodeIDs(nodes)))
-
-	r.sendSegments(seqRequests)
-
-	// next time we send requests, we start with a different permutation
-	r.requestIterator++
+	r.sendRequests(seqsOrRounds)
 }
 
-func (r *requestor) sendSegments(segments []Segment) {
-	numNodes := len(r.highestObserved.signers)
-	for i, seqsOrRounds := range segments {
+// sendRequests requests [seqsOrRounds] from the nodes that have signed
+// [highestObserved], by splitting them into batches of at most
+// MaxRoundRequests and sending each batch to one of the signers.
+// It is the single send path, used both for initial replication requests
+// and for re-sending requests that have timed out.
+func (r *requestor) sendRequests(seqsOrRounds []uint64) {
+	signers := r.highestObserved.signers
+	numNodes := len(signers)
+	batches := BatchSequences(seqsOrRounds, uint64(numNodes), maxItemCountPerRequest)
+
+	for i, batch := range batches {
 		index := (i + r.requestIterator) % numNodes
-		r.sendRequestToNode(seqsOrRounds.Start, seqsOrRounds.End, r.highestObserved.signers[index])
+		r.sendRequestToNode(batch, signers[index])
 	}
+	r.requestIterator++
 }
 
-// sendRequestToNode requests [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 *requestor) sendRequestToNode(start uint64, end uint64, node common.NodeID) {
-	seqsOrRound := make([]uint64, 0, (end+1)-start)
-	for i := start; i <= end; i++ {
+// sendRequestToNode requests [seqsOrRounds] from node, skipping
+// any sequences we have already committed. In case the node does not respond,
+// we create a timeout that will re-send the request. seqsOrRounds is expected to be sorted in ascending order,
+// as the last element is used to update highestRequested.
+func (r *requestor) sendRequestToNode(seqsOrRounds []uint64, node common.NodeID) {
+	toRequest := make([]uint64, 0, len(seqsOrRounds))
+	for _, seqOrRound := range seqsOrRounds {
 		// Skip sequences we have already committed;
-		if r.replicateSeqs && r.highestCommitted != nil && i <= *r.highestCommitted {
+		if r.replicateSeqs && r.highestCommitted != nil && seqOrRound <= *r.highestCommitted {
 			continue
 		}
-		seqsOrRound = append(seqsOrRound, i)
+		toRequest = append(toRequest, seqOrRound)
 		// ensure we set a timeout for this sequence
-		r.timeoutHandler.AddTask(i)
+		r.timeoutHandler.AddTask(seqOrRound)
 	}
 
-	if len(seqsOrRound) == 0 {
+	if len(toRequest) == 0 {
 		return
 	}
 
-	if r.highestRequested < end {
-		r.highestRequested = end
+	if !slices.IsSorted(toRequest) {
+		slices.Sort(toRequest)
+	}
+	if last := toRequest[len(toRequest)-1]; last > r.highestRequested {
+		r.highestRequested = last
 	}
 
 	request := &common.ReplicationRequest{}
 	if r.replicateSeqs {
 		request.LatestFinalizedSeq = r.highestObserved.seq
-		request.Seqs = seqsOrRound
+		request.Seqs = toRequest
 	} else {
 		request.LatestRound = r.highestObserved.round
-		request.Rounds = seqsOrRound
+		request.Rounds = toRequest
 	}
 
 	msg := &common.Message{ReplicationRequest: request}
 
 	r.logger.Debug("Requesting missing rounds/sequences ",
 		zap.Stringer("from", node),
-		zap.Uint64("start", start),
-		zap.Uint64("end", end),
 		zap.Int("sequence count", len(request.Seqs)),
 		zap.Int("round count", len(request.Rounds)),
 		zap.Uint64("latestSeq", request.LatestFinalizedSeq),

simplex/util.go

@@ -185,81 +185,24 @@ func (block *oneTimeVerifiedBlock) Verify(ctx context.Context) (common.VerifiedB
 	return vb, err
 }
 
-type Segment struct {
-	Start uint64
-	End   uint64
-}
-
-// compressSequences takes a slice of uint64 values representing
-// missing sequence numbers and compresses consecutive numbers into segments.
-// Each segment represents a continuous block of missing sequence numbers.
-func CompressSequences(missingSeqs []uint64) []Segment {
-	slices.Sort(missingSeqs)
-	var segments []Segment
-
-	if len(missingSeqs) == 0 {
-		return segments
+// BatchSequences distributes [seqs] as evenly as possible among [numNodes]
+// nodes, returning one batch per node share. Every batch contains at most
+// [maxSize] sequences. a share exceeding [maxSize] is emitted as multiple
+// batches. [seqs] does not need to be sorted or contiguous, it is sorted in
+// place, and the returned batches are sub-slices of it, ordered from lowest
+// to highest sequence.
+func BatchSequences(seqs []uint64, numNodes uint64, maxSize uint64) [][]uint64 {
+	if len(seqs) == 0 || numNodes == 0 || maxSize == 0 {
+		return nil
 	}
+	slices.Sort(seqs)
 
-	startSeq := missingSeqs[0]
-	endSeq := missingSeqs[0]
-
-	for i, currentSeq := range missingSeqs[1:] {
-		if currentSeq != missingSeqs[i]+1 {
-			segments = append(segments, Segment{
-				Start: startSeq,
-				End:   endSeq,
-			})
-			startSeq = currentSeq
-		}
-		endSeq = currentSeq
-	}
+	numSeqs := uint64(len(seqs))
 
-	segments = append(segments, Segment{
-		Start: startSeq,
-		End:   endSeq,
-	})
-
-	return segments
-}
-
-// DistributeSequenceRequests evenly creates segments amongst [numNodes] over
-// the range [start, end].
-func DistributeSequenceRequests(start, end uint64, numNodes int) []Segment {
-	var segments []Segment
-
-	if numNodes <= 0 || start > end {
-		return segments
-	}
-
-	numSeqs := end + 1 - start
-	seqsPerNode := numSeqs / uint64(numNodes)
-	remainder := numSeqs % uint64(numNodes)
-
-	if seqsPerNode == 0 {
-		seqsPerNode = 1
-	}
-
-	nodeStart := start
-
-	for i := 0; i < numNodes && nodeStart <= end; i++ {
-		segmentLength := seqsPerNode
-		if remainder > 0 {
-			segmentLength++
-			remainder--
-		}
-
-		nodeEnd := min(nodeStart+segmentLength-1, end)
-
-		segments = append(segments, Segment{
-			Start: nodeStart,
-			End:   nodeEnd,
-		})
-
-		nodeStart = nodeEnd + 1
-	}
+	seqsPerNode := (numSeqs + uint64(numNodes) - 1) / uint64(numNodes)
 
-	return segments
+	batchSize := min(seqsPerNode, maxSize)
+	return slices.Collect(slices.Chunk(seqs, int(batchSize)))
 }
 
 type NotarizationTime struct {