fix replication requests exceeding maxRoundWindow + splitting resends

simplex/replication_test.go

@@ -1760,3 +1760,50 @@ messages:
 	notarization := wal.AssertNotarization(uint64(len(blocks)))
 	require.Equal(t, common.EmptyNotarizationRecordType, notarization)
 }
+
+// TestReplicationRequestsWithinMaxRoundWindow ensures that a node that observes a finalization more than MaxRoundWindow
+// sequences ahead of its next sequence to commit only requests sequences up to MaxRoundWindow-1 ahead of it
+func TestReplicationRequestsWithinMaxRoundWindow(t *testing.T) {
+	bb := testutil.NewTestBlockBuilder()
+	nodes := []common.NodeID{{1}, {2}, {3}, {4}}
+	sentMessages := make(chan *common.Message, 100)
+	conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[1], &recordingComm{
+		Communication: testutil.NewNoopComm(nodes),
+		SentMessages:  sentMessages,
+	}, bb)
+	conf.ReplicationEnabled = true
+
+	e, err := simplex.NewEpoch(conf)
+	require.NoError(t, err)
+	t.Cleanup(e.Stop)
+	require.NoError(t, e.Start())
+
+	// observe a finalization more than MaxRoundWindow sequences ahead of us
+	seqCount := 2 * conf.MaxRoundWindow
+	finalization := createBlocks(t, nodes, seqCount)[seqCount-1].Finalization
+	err = e.HandleMessage(&common.Message{Finalization: &finalization}, nodes[0])
+	require.NoError(t, err)
+
+	// collect the replication requests sent out in response to the finalization
+	requested := make(map[uint64]struct{})
+	timeout := time.After(30 * time.Second)
+	for uint64(len(requested)) < conf.MaxRoundWindow {
+		select {
+		case msg := <-sentMessages:
+			if msg.ReplicationRequest == nil {
+				continue
+			}
+			for _, seq := range msg.ReplicationRequest.Seqs {
+				requested[seq] = struct{}{}
+			}
+		case <-timeout:
+			require.FailNow(t, "timed out waiting for replication requests")
+		}
+	}
+
+	// our next sequence to commit is 0, so we may request at most sequences [0, MaxRoundWindow-1]
+	for seq := range requested {
+		require.Less(t, seq, conf.MaxRoundWindow,
+			"requested a sequence more than MaxRoundWindow ahead of the next sequence to commit")
+	}
+}

simplex/replication_timeout_test.go

@@ -677,3 +677,70 @@ func TestReplicationResendsFinalizedBlocksThatFailedVerification(t *testing.T) {
 	require.Equal(t, uint64(1), storage.NumBlocks())
 	require.Equal(t, block, storedBlock)
 }
+
+// TestReplicationResendSplitsRequests 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.
+func TestReplicationResendSplitsRequests(t *testing.T) {
+	bb := testutil.NewTestBlockBuilder()
+	nodes := []common.NodeID{{1}, {2}, {3}, {4}}
+	sentMessages := make(chan *common.Message, 100)
+	conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[1], &recordingComm{
+		Communication: testutil.NewNoopComm(nodes),
+		SentMessages:  sentMessages,
+	}, bb)
+	conf.ReplicationEnabled = true
+
+	e, err := simplex.NewEpoch(conf)
+	require.NoError(t, err)
+	t.Cleanup(e.Stop)
+	require.NoError(t, e.Start())
+
+	// observe a finalization more than MaxRoundWindow sequences ahead of us
+	seqCount := 2 * conf.MaxRoundWindow
+	finalization := createBlocks(t, nodes, seqCount)[seqCount-1].Finalization
+	err = e.HandleMessage(&common.Message{Finalization: &finalization}, nodes[0])
+	require.NoError(t, err)
+
+	// collect the initial replication requests sent in response to the finalization
+	initiallyRequested := make(map[uint64]struct{})
+	timeout := time.After(30 * time.Second)
+	for uint64(len(initiallyRequested)) < conf.MaxRoundWindow {
+		select {
+		case msg := <-sentMessages:
+			if msg.ReplicationRequest == nil {
+				continue
+			}
+			for _, seq := range msg.ReplicationRequest.Seqs {
+				initiallyRequested[seq] = struct{}{}
+			}
+		case <-timeout:
+			require.FailNow(t, "timed out waiting for replication requests")
+		}
+	}
+
+	// no node responds, so the requests time out and are re-sent
+	e.AdvanceTime(e.StartTime.Add(2 * simplex.DefaultReplicationRequestTimeout))
+
+	// collect the re-sent requests until they cover all outstanding sequences
+	resentRequests := 0
+	resentSeqs := make(map[uint64]struct{})
+	for len(resentSeqs) < len(initiallyRequested) {
+		select {
+		case msg := <-sentMessages:
+			if msg.ReplicationRequest == nil {
+				continue
+			}
+			resentRequests++
+			require.LessOrEqual(t, uint64(len(msg.ReplicationRequest.Seqs)), conf.MaxRoundWindow,
+				"a replication request with more than MaxRoundWindow seqs is dropped by the responder")
+			for _, seq := range msg.ReplicationRequest.Seqs {
+				resentSeqs[seq] = struct{}{}
+			}
+		case <-timeout:
+			require.FailNow(t, "timed out waiting for re-sent replication requests")
+		}
+	}
+
+	require.Greater(t, resentRequests, 1,
+		"timed out requests should be re-sent split across the signers of the highest observed quorum")
+}

simplex/requestor.go

@@ -128,7 +128,15 @@ func (r *requestor) resendReplicationRequests(missingIds []uint64) {
 	r.epochLock.Lock()
 	defer r.epochLock.Unlock()
 
-	segments := CompressSequences(missingIds)
+	numNodes := len(r.highestObserved.signers)
+
+	// split each contiguous range among the nodes that signed,
+	// just like the initial requests, so that a single request never exceeds
+	// the maxRoundWindow limit enforced by the responding nodes
+	var segments []Segment
+	for _, segment := range CompressSequences(missingIds) {
+		segments = append(segments, DistributeSequenceRequests(segment.Start, segment.End, numNodes)...)
+	}
 
 	r.sendSegments(segments)
 
@@ -159,7 +167,7 @@ func (r *requestor) observedSignedQuorum(observed *signedQuorum, currentSeqOrRou
 func (r *requestor) sendMoreReplicationRequests(observedSeqOrRound, currentSeqOrRound uint64) {
 	start := 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))
+	end := math.Min(float64(observedSeqOrRound), float64(r.maxRoundWindow+currentSeqOrRound-1))
 
 	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))