diff --git a/simplex/replication_test.go b/simplex/replication_test.go index 3c155510..0a700cf9 100644 --- a/simplex/replication_test.go +++ b/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") + } +} diff --git a/simplex/replication_timeout_test.go b/simplex/replication_timeout_test.go index 9b741d8f..0c2f619c 100644 --- a/simplex/replication_timeout_test.go +++ b/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") +} diff --git a/simplex/requestor.go b/simplex/requestor.go index ac1f5d70..806db29d 100644 --- a/simplex/requestor.go +++ b/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))