From ba41f4e73a8e0b18d03a394d2b9dfa9fbef982d8 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Thu, 2 Jul 2026 16:55:46 -0400 Subject: [PATCH 1/8] fix replication requests exceeding maxRoundWindow and splitting resendRequests --- simplex/replication_test.go | 47 ++++++++++++++++++++ simplex/replication_timeout_test.go | 67 +++++++++++++++++++++++++++++ simplex/requestor.go | 12 +++++- 3 files changed, 124 insertions(+), 2 deletions(-) diff --git a/simplex/replication_test.go b/simplex/replication_test.go index 3c155510..e59dd8ae 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..60fb76fa 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..6241b9a3 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)) From f45ac2b4cbfbf08b897307f599011c7ff62471ff Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Thu, 2 Jul 2026 17:05:50 -0400 Subject: [PATCH 2/8] go format fixed --- simplex/replication_test.go | 2 +- simplex/replication_timeout_test.go | 2 +- simplex/requestor.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/simplex/replication_test.go b/simplex/replication_test.go index e59dd8ae..0a700cf9 100644 --- a/simplex/replication_test.go +++ b/simplex/replication_test.go @@ -1761,7 +1761,7 @@ messages: require.Equal(t, common.EmptyNotarizationRecordType, notarization) } -// TestReplicationRequestsWithinMaxRoundWindow ensures that a node that observes a finalization more than MaxRoundWindow +// 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() diff --git a/simplex/replication_timeout_test.go b/simplex/replication_timeout_test.go index 60fb76fa..0c2f619c 100644 --- a/simplex/replication_timeout_test.go +++ b/simplex/replication_timeout_test.go @@ -678,7 +678,7 @@ func TestReplicationResendsFinalizedBlocksThatFailedVerification(t *testing.T) { require.Equal(t, block, storedBlock) } -// TestReplicationResendSplitsRequests ensures that when replication requests time out, the missing sequences +// 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() diff --git a/simplex/requestor.go b/simplex/requestor.go index 6241b9a3..806db29d 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -167,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 - 1)) + 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)) From 06bbf59fb1da21e5331bd00032ac437a6c55ce10 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Mon, 6 Jul 2026 17:34:02 -0400 Subject: [PATCH 3/8] extract DistributeMissingSequences and enfoce maxRoundWindow on retried requests --- .gitignore | 2 +- simplex/replication_test.go | 49 +--------------- simplex/requestor.go | 8 +-- simplex/util.go | 17 ++++++ simplex/util_test.go | 109 ++++++++++++++++++++++++++++++++++++ 5 files changed, 130 insertions(+), 55 deletions(-) diff --git a/.gitignore b/.gitignore index 55892921..81837034 100644 --- a/.gitignore +++ b/.gitignore @@ -32,4 +32,4 @@ go.work.sum .idea/ # tmp folder -/tmp \ No newline at end of file +/tmptmp/ diff --git a/simplex/replication_test.go b/simplex/replication_test.go index 0a700cf9..a25851cb 100644 --- a/simplex/replication_test.go +++ b/simplex/replication_test.go @@ -1759,51 +1759,4 @@ messages: require.True(t, foundEmptyVote, "Node should send an empty vote after timeout following replication") 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") - } -} +} \ No newline at end of file diff --git a/simplex/requestor.go b/simplex/requestor.go index 806db29d..ebbe5148 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -128,15 +128,11 @@ func (r *requestor) resendReplicationRequests(missingIds []uint64) { r.epochLock.Lock() defer r.epochLock.Unlock() - 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)...) - } + segments := DistributeMissingSequences(missingIds, len(r.highestObserved.signers), r.maxRoundWindow) r.sendSegments(segments) @@ -167,7 +163,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-1)) + 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)) diff --git a/simplex/util.go b/simplex/util.go index e59d96cd..572ed2d9 100644 --- a/simplex/util.go +++ b/simplex/util.go @@ -262,6 +262,23 @@ func DistributeSequenceRequests(start, end uint64, numNodes int) []Segment { return segments } + +func DistributeMissingSequences(missingSeqs []uint64, numNodes int, maxSize uint64) []Segment { + var segments []Segment + if maxSize == 0 { + return nil + } + for _, segment := range CompressSequences(missingSeqs) { + for _, distributed := range DistributeSequenceRequests(segment.Start, segment.End, numNodes) { + for start := distributed.Start; start <= distributed.End; start+= maxSize { + segments = append(segments, Segment {Start: start, + End: min (start + maxSize - 1, distributed.End)}) + } + } + } + return segments +} + type NotarizationTime struct { // config getRound func() uint64 diff --git a/simplex/util_test.go b/simplex/util_test.go index 4513fc78..26907474 100644 --- a/simplex/util_test.go +++ b/simplex/util_test.go @@ -378,6 +378,115 @@ func TestDistributeSequenceRequests(t *testing.T) { } } +func TestDistributeMissingSequences(t *testing.T) { + tests := []struct { + name string + missingSeqs []uint64 + numNodes int + maxSize uint64 + expected []Segment + }{ { + name: "empty input", + missingSeqs: []uint64{}, + numNodes: 3, + maxSize: 10, + expected: nil, + }, + { + name: "single missing sequence", + missingSeqs: []uint64{5}, + numNodes: 3, + maxSize: 10, + expected: []Segment{{Start: 5, End: 5}}, + }, + { + name: "contiguous range split among nodes", + missingSeqs: []uint64{5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}, + numNodes: 3, + maxSize: 10, + expected: []Segment{ + {Start: 5, End: 8}, + {Start: 9, End: 12}, + {Start: 13, End: 15}, + }, + }, + { + name: "unsorted input with gaps", + missingSeqs: []uint64{9,7,3,0,1,2,8}, + numNodes: 2, + maxSize: 10, + expected: []Segment{ + {Start: 0, End: 1}, + {Start: 2, End: 3}, + {Start: 7, End: 8}, + {Start: 9, End: 9}, + }, + }, + { + name: "single node segments capped at maxSize", + missingSeqs: []uint64{ + 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, + 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, + }, + numNodes: 1, + maxSize: 10, + expected: []Segment{ + {Start: 0, End: 9}, + {Start: 10, End: 19}, + {Start: 20, End: 24}, + }, + }, + { + name: "zero max size", + missingSeqs: []uint64{1,2,3}, + numNodes: 2, + maxSize: 0, + expected: nil, + }, + { + name: "zero nodes", + missingSeqs: []uint64{1,2,3}, + numNodes: 0, + maxSize: 10, + expected: nil, + }, + { + name: "range exceeds maxSize but node shares are within it", + missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, + numNodes: 2, + maxSize: 10, + expected: []Segment{ + {Start: 0, End: 7}, + {Start: 8, End: 14}, + }, + }, + { + name: "node shares still exceed maxSize and are chunked", + missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, + numNodes: 3, + maxSize: 3, + expected: []Segment{ + {Start: 0, End: 2}, + {Start: 3, End: 4}, + {Start: 5, End: 7}, + {Start: 8, End: 9}, + {Start: 10, End: 12}, + {Start: 13, End: 14}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := DistributeMissingSequences(tt.missingSeqs, tt.numNodes, tt.maxSize) + require.Equal(t, tt.expected, result) + }) + } + +} + + + func TestNotarizationTime(t *testing.T) { defaultFinalizeVoteRebroadcastTimeout := time.Second * 6 From 3f1ec1f104dec1982757cf96eeda58f8bf44d117 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Mon, 6 Jul 2026 17:57:29 -0400 Subject: [PATCH 4/8] fix .gitignore rule and formatting --- .gitignore | 2 +- simplex/replication_test.go | 2 +- simplex/requestor.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index 81837034..ccec156b 100644 --- a/.gitignore +++ b/.gitignore @@ -32,4 +32,4 @@ go.work.sum .idea/ # tmp folder -/tmptmp/ +/tmp diff --git a/simplex/replication_test.go b/simplex/replication_test.go index a25851cb..3c155510 100644 --- a/simplex/replication_test.go +++ b/simplex/replication_test.go @@ -1759,4 +1759,4 @@ messages: require.True(t, foundEmptyVote, "Node should send an empty vote after timeout following replication") notarization := wal.AssertNotarization(uint64(len(blocks))) require.Equal(t, common.EmptyNotarizationRecordType, notarization) -} \ No newline at end of file +} diff --git a/simplex/requestor.go b/simplex/requestor.go index ebbe5148..eab493e6 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -163,7 +163,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 -1)) + 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)) From 12f19ea1bb3047066d52474036c542f12310ee12 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Mon, 6 Jul 2026 18:03:28 -0400 Subject: [PATCH 5/8] format --- simplex/requestor.go | 1 - simplex/util.go | 7 +- simplex/util_test.go | 178 +++++++++++++++++++++---------------------- 3 files changed, 91 insertions(+), 95 deletions(-) diff --git a/simplex/requestor.go b/simplex/requestor.go index eab493e6..907e2df5 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -128,7 +128,6 @@ func (r *requestor) resendReplicationRequests(missingIds []uint64) { r.epochLock.Lock() defer r.epochLock.Unlock() - // 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 diff --git a/simplex/util.go b/simplex/util.go index 572ed2d9..4658dad0 100644 --- a/simplex/util.go +++ b/simplex/util.go @@ -262,7 +262,6 @@ func DistributeSequenceRequests(start, end uint64, numNodes int) []Segment { return segments } - func DistributeMissingSequences(missingSeqs []uint64, numNodes int, maxSize uint64) []Segment { var segments []Segment if maxSize == 0 { @@ -270,9 +269,9 @@ func DistributeMissingSequences(missingSeqs []uint64, numNodes int, maxSize uint } for _, segment := range CompressSequences(missingSeqs) { for _, distributed := range DistributeSequenceRequests(segment.Start, segment.End, numNodes) { - for start := distributed.Start; start <= distributed.End; start+= maxSize { - segments = append(segments, Segment {Start: start, - End: min (start + maxSize - 1, distributed.End)}) + for start := distributed.Start; start <= distributed.End; start += maxSize { + segments = append(segments, Segment{Start: start, + End: min(start+maxSize-1, distributed.End)}) } } } diff --git a/simplex/util_test.go b/simplex/util_test.go index 26907474..00cc41f6 100644 --- a/simplex/util_test.go +++ b/simplex/util_test.go @@ -380,100 +380,100 @@ func TestDistributeSequenceRequests(t *testing.T) { func TestDistributeMissingSequences(t *testing.T) { tests := []struct { - name string + name string missingSeqs []uint64 - numNodes int - maxSize uint64 - expected []Segment - }{ { - name: "empty input", - missingSeqs: []uint64{}, - numNodes: 3, - maxSize: 10, - expected: nil, - }, - { - name: "single missing sequence", - missingSeqs: []uint64{5}, - numNodes: 3, - maxSize: 10, - expected: []Segment{{Start: 5, End: 5}}, - }, - { - name: "contiguous range split among nodes", - missingSeqs: []uint64{5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}, - numNodes: 3, - maxSize: 10, - expected: []Segment{ - {Start: 5, End: 8}, - {Start: 9, End: 12}, - {Start: 13, End: 15}, - }, - }, - { - name: "unsorted input with gaps", - missingSeqs: []uint64{9,7,3,0,1,2,8}, - numNodes: 2, - maxSize: 10, - expected: []Segment{ - {Start: 0, End: 1}, - {Start: 2, End: 3}, - {Start: 7, End: 8}, - {Start: 9, End: 9}, - }, - }, - { - name: "single node segments capped at maxSize", - missingSeqs: []uint64{ - 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, - 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, - }, - numNodes: 1, - maxSize: 10, - expected: []Segment{ - {Start: 0, End: 9}, - {Start: 10, End: 19}, - {Start: 20, End: 24}, - }, - }, - { - name: "zero max size", - missingSeqs: []uint64{1,2,3}, - numNodes: 2, - maxSize: 0, - expected: nil, - }, - { - name: "zero nodes", - missingSeqs: []uint64{1,2,3}, - numNodes: 0, - maxSize: 10, - expected: nil, - }, - { - name: "range exceeds maxSize but node shares are within it", - missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, - numNodes: 2, - maxSize: 10, - expected: []Segment{ + numNodes int + maxSize uint64 + expected []Segment + }{{ + name: "empty input", + missingSeqs: []uint64{}, + numNodes: 3, + maxSize: 10, + expected: nil, + }, + { + name: "single missing sequence", + missingSeqs: []uint64{5}, + numNodes: 3, + maxSize: 10, + expected: []Segment{{Start: 5, End: 5}}, + }, + { + name: "contiguous range split among nodes", + missingSeqs: []uint64{5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}, + numNodes: 3, + maxSize: 10, + expected: []Segment{ + {Start: 5, End: 8}, + {Start: 9, End: 12}, + {Start: 13, End: 15}, + }, + }, + { + name: "unsorted input with gaps", + missingSeqs: []uint64{9, 7, 3, 0, 1, 2, 8}, + numNodes: 2, + maxSize: 10, + expected: []Segment{ + {Start: 0, End: 1}, + {Start: 2, End: 3}, + {Start: 7, End: 8}, + {Start: 9, End: 9}, + }, + }, + { + name: "single node segments capped at maxSize", + missingSeqs: []uint64{ + 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, + 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, + }, + numNodes: 1, + maxSize: 10, + expected: []Segment{ + {Start: 0, End: 9}, + {Start: 10, End: 19}, + {Start: 20, End: 24}, + }, + }, + { + name: "zero max size", + missingSeqs: []uint64{1, 2, 3}, + numNodes: 2, + maxSize: 0, + expected: nil, + }, + { + name: "zero nodes", + missingSeqs: []uint64{1, 2, 3}, + numNodes: 0, + maxSize: 10, + expected: nil, + }, + { + name: "range exceeds maxSize but node shares are within it", + missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, + numNodes: 2, + maxSize: 10, + expected: []Segment{ {Start: 0, End: 7}, {Start: 8, End: 14}, - }, }, + }, { - name: "node shares still exceed maxSize and are chunked", - missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, - numNodes: 3, - maxSize: 3, - expected: []Segment{ - {Start: 0, End: 2}, - {Start: 3, End: 4}, - {Start: 5, End: 7}, - {Start: 8, End: 9}, - {Start: 10, End: 12}, - {Start: 13, End: 14}, - }, + name: "node shares still exceed maxSize and are chunked", + missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, + numNodes: 3, + maxSize: 3, + expected: []Segment{ + {Start: 0, End: 2}, + {Start: 3, End: 4}, + {Start: 5, End: 7}, + {Start: 8, End: 9}, + {Start: 10, End: 12}, + {Start: 13, End: 14}, }, + }, } for _, tt := range tests { @@ -485,8 +485,6 @@ func TestDistributeMissingSequences(t *testing.T) { } - - func TestNotarizationTime(t *testing.T) { defaultFinalizeVoteRebroadcastTimeout := time.Second * 6 From 6a8d1c2973f45470478bccae0dc1e5c38f8c3622 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Wed, 8 Jul 2026 15:31:16 -0400 Subject: [PATCH 6/8] add batchSequences -- initial and retry replication paths call the same function --- simplex/epoch.go | 1 + simplex/replication_state.go | 2 +- simplex/requestor.go | 72 ++++------ simplex/util.go | 102 +++---------- simplex/util_test.go | 271 +++++++++++------------------------ 5 files changed, 130 insertions(+), 318 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index 13b10f70..51f5d831 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -30,6 +30,7 @@ const ( DefaultEmptyVoteRebroadcastTimeout = 5 * time.Second DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second EmptyVoteTimeoutID = "rebroadcast_empty_vote" + MaxRoundRequests = 10 ) type EmptyVoteSet struct { diff --git a/simplex/replication_state.go b/simplex/replication_state.go index 89443388..ed9e2a28 100644 --- a/simplex/replication_state.go +++ b/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 diff --git a/simplex/requestor.go b/simplex/requestor.go index 907e2df5..0eaba228 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -128,14 +128,7 @@ func (r *requestor) resendReplicationRequests(missingIds []uint64) { r.epochLock.Lock() defer r.epochLock.Unlock() - // 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 - segments := DistributeMissingSequences(missingIds, len(r.highestObserved.signers), r.maxRoundWindow) - - r.sendSegments(segments) - - r.requestIterator++ + r.sendRequests(missingIds) } // observedSignedQuorum is called when we observe a signed quorum for a future round/sequence. @@ -160,76 +153,69 @@ 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)) + r.sendRequests(seqsOrRounds) } -// 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) +func (r *requestor) sendRequests(seqsOrRounds []uint64) { + signers := r.highestObserved.signers + numNodes := len(signers) + batches := BatchSequences(seqsOrRounds, numNodes, MaxRoundRequests) - 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++ -} - -func (r *requestor) sendSegments(segments []Segment) { - numNodes := len(r.highestObserved.signers) - for i, seqsOrRounds := range segments { + 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++ { +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 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), diff --git a/simplex/util.go b/simplex/util.go index 4658dad0..556aea06 100644 --- a/simplex/util.go +++ b/simplex/util.go @@ -185,97 +185,31 @@ 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 - } - - 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 - } - - 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 +func BatchSequences(seqs []uint64, numNodes int, maxSize uint64) [][]uint64 { + if len(seqs) == 0 || numNodes <= 0 || maxSize == 0 { + return nil } + slices.Sort(seqs) - numSeqs := end + 1 - start - seqsPerNode := numSeqs / uint64(numNodes) + numSeqs := uint64(len(seqs)) + share := 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-- + var batches [][]uint64 + start := uint64(0) + for i := 0; i < numNodes && start < numSeqs; i++ { + nodeShare := share + if uint64(i) < remainder { + nodeShare++ } - - nodeEnd := min(nodeStart+segmentLength-1, end) - - segments = append(segments, Segment{ - Start: nodeStart, - End: nodeEnd, - }) - - nodeStart = nodeEnd + 1 - } - - return segments -} - -func DistributeMissingSequences(missingSeqs []uint64, numNodes int, maxSize uint64) []Segment { - var segments []Segment - if maxSize == 0 { - return nil - } - for _, segment := range CompressSequences(missingSeqs) { - for _, distributed := range DistributeSequenceRequests(segment.Start, segment.End, numNodes) { - for start := distributed.Start; start <= distributed.End; start += maxSize { - segments = append(segments, Segment{Start: start, - End: min(start+maxSize-1, distributed.End)}) - } + for nodeShare > 0 { + n := min(nodeShare, maxSize) + batches = append(batches, seqs[start:start+n]) + start += n + nodeShare -= n } } - return segments + return batches } type NotarizationTime struct { diff --git a/simplex/util_test.go b/simplex/util_test.go index 00cc41f6..201d2f14 100644 --- a/simplex/util_test.go +++ b/simplex/util_test.go @@ -242,247 +242,138 @@ func TestGetHighestQuorumRound(t *testing.T) { }) } } - -func TestCompressSequences(t *testing.T) { +func TestBatchSequences(t *testing.T) { tests := []struct { name string - input []uint64 - expected []Segment + seqs []uint64 + numNodes int + maxSize uint64 + expected [][]uint64 }{ { name: "empty input", - input: []uint64{}, + seqs: []uint64{}, + numNodes: 3, + maxSize: 10, expected: nil, }, { - name: "single element", - input: []uint64{5}, - expected: []Segment{ - {Start: 5, End: 5}, - }, - }, - { - name: "all consecutive", - input: []uint64{1, 2, 3, 4, 5}, - expected: []Segment{ - {Start: 1, End: 5}, - }, + name: "zero nodes", + seqs: []uint64{1, 2, 3}, + numNodes: 0, + maxSize: 10, + expected: nil, }, { - name: "no consecutive elements", - input: []uint64{2, 4, 6, 8}, - expected: []Segment{ - {Start: 2, End: 2}, - {Start: 4, End: 4}, - {Start: 6, End: 6}, - {Start: 8, End: 8}, - }, + name: "zero max size", + seqs: []uint64{1, 2, 3}, + numNodes: 2, + maxSize: 0, + expected: nil, }, { - name: "mixed consecutive and non-consecutive", - input: []uint64{3, 4, 5, 7, 8, 10}, - expected: []Segment{ - {Start: 3, End: 5}, - {Start: 7, End: 8}, - {Start: 10, End: 10}, - }, + name: "single sequence", + seqs: []uint64{5}, + numNodes: 3, + maxSize: 10, + expected: [][]uint64{{5}}, }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := CompressSequences(tt.input) - require.Equal(t, tt.expected, result) - }) - } -} - -func TestDistributeSequenceRequests(t *testing.T) { - tests := []struct { - name string - start uint64 - end uint64 - numNodes int - expected []Segment - }{ { - name: "even distribution", - start: 0, - end: 9, - numNodes: 2, - expected: []Segment{ - {Start: 0, End: 4}, - {Start: 5, End: 9}, + name: "even split among nodes", + seqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8}, + numNodes: 3, + maxSize: 10, + expected: [][]uint64{ + {0, 1, 2}, + {3, 4, 5}, + {6, 7, 8}, }, }, { - name: "uneven distribution", - start: 0, - end: 10, + name: "even split among nodes with gaps", + seqs: []uint64{0, 1, 3, 4, 6, 7, 8, 10, 11}, numNodes: 3, - expected: []Segment{ - {Start: 0, End: 3}, - {Start: 4, End: 7}, - {Start: 8, End: 10}, + maxSize: 10, + expected: [][]uint64{ + {0, 1, 3}, + {4, 6, 7}, + {8, 10, 11}, }, }, { - name: "single node full range", - start: 5, - end: 15, - numNodes: 1, - expected: []Segment{ - {Start: 5, End: 15}, + name: "remainder goes to first nodes", + seqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, + numNodes: 3, + maxSize: 10, + expected: [][]uint64{ + {0, 1, 2, 3}, + {4, 5, 6}, + {7, 8, 9}, }, }, { - name: "numNodes greater than sequences", - start: 0, - end: 2, + name: "gaps split by total not per run", + seqs: []uint64{0, 1, 2, 3, 4, 6, 7, 8, 9, 10, 12, 13, 14, 15}, numNodes: 5, - expected: []Segment{ - {Start: 0, End: 1}, - {Start: 2, End: 2}, + maxSize: 40, + expected: [][]uint64{ + {0, 1, 2}, + {3, 4, 6}, + {7, 8, 9}, + {10, 12, 13}, + {14, 15}, }, }, { - name: "zero-length range", - start: 5, - end: 5, - numNodes: 3, - expected: []Segment{ - {Start: 5, End: 5}, - }, - }, - { - name: "start > end", - start: 10, - end: 5, + name: "unsorted input is sorted first", + seqs: []uint64{9, 0, 4, 2, 7}, numNodes: 2, - expected: nil, - }, - { - name: "zero nodes", - start: 0, - end: 10, - numNodes: 0, - expected: nil, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := DistributeSequenceRequests(tt.start, tt.end, tt.numNodes) - require.Equal(t, tt.expected, result) - }) - } -} - -func TestDistributeMissingSequences(t *testing.T) { - tests := []struct { - name string - missingSeqs []uint64 - numNodes int - maxSize uint64 - expected []Segment - }{{ - name: "empty input", - missingSeqs: []uint64{}, - numNodes: 3, - maxSize: 10, - expected: nil, - }, - { - name: "single missing sequence", - missingSeqs: []uint64{5}, - numNodes: 3, - maxSize: 10, - expected: []Segment{{Start: 5, End: 5}}, - }, - { - name: "contiguous range split among nodes", - missingSeqs: []uint64{5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}, - numNodes: 3, - maxSize: 10, - expected: []Segment{ - {Start: 5, End: 8}, - {Start: 9, End: 12}, - {Start: 13, End: 15}, + maxSize: 10, + expected: [][]uint64{ + {0, 2, 4}, + {7, 9}, }, }, { - name: "unsorted input with gaps", - missingSeqs: []uint64{9, 7, 3, 0, 1, 2, 8}, - numNodes: 2, - maxSize: 10, - expected: []Segment{ - {Start: 0, End: 1}, - {Start: 2, End: 3}, - {Start: 7, End: 8}, - {Start: 9, End: 9}, - }, + name: "more nodes than sequences", + seqs: []uint64{1, 2, 3}, + numNodes: 5, + maxSize: 10, + expected: [][]uint64{{1}, {2}, {3}}, }, { - name: "single node segments capped at maxSize", - missingSeqs: []uint64{ + name: "single node share capped at maxSize", + seqs: []uint64{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, }, numNodes: 1, maxSize: 10, - expected: []Segment{ - {Start: 0, End: 9}, - {Start: 10, End: 19}, - {Start: 20, End: 24}, + expected: [][]uint64{ + {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, + {10, 11, 12, 13, 14, 15, 16, 17, 18, 19}, + {20, 21, 22, 23, 24}, }, }, { - name: "zero max size", - missingSeqs: []uint64{1, 2, 3}, - numNodes: 2, - maxSize: 0, - expected: nil, - }, - { - name: "zero nodes", - missingSeqs: []uint64{1, 2, 3}, - numNodes: 0, - maxSize: 10, - expected: nil, - }, - { - name: "range exceeds maxSize but node shares are within it", - missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, - numNodes: 2, - maxSize: 10, - expected: []Segment{ - {Start: 0, End: 7}, - {Start: 8, End: 14}, - }, - }, - { - name: "node shares still exceed maxSize and are chunked", - missingSeqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, - numNodes: 3, - maxSize: 3, - expected: []Segment{ - {Start: 0, End: 2}, - {Start: 3, End: 4}, - {Start: 5, End: 7}, - {Start: 8, End: 9}, - {Start: 10, End: 12}, - {Start: 13, End: 14}, + name: "node shares exceed maxSize and are chunked", + seqs: []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}, + numNodes: 3, + maxSize: 3, + expected: [][]uint64{ + {0, 1, 2}, {3, 4}, + {5, 6, 7}, {8, 9}, + {10, 11, 12}, {13, 14}, }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result := DistributeMissingSequences(tt.missingSeqs, tt.numNodes, tt.maxSize) + result := BatchSequences(tt.seqs, tt.numNodes, tt.maxSize) require.Equal(t, tt.expected, result) }) } - } func TestNotarizationTime(t *testing.T) { From 523180aed5883b3641e3d7d6663fb7ade3b9d4ae Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Wed, 8 Jul 2026 17:49:46 -0400 Subject: [PATCH 7/8] add comments + simplify code to use Slice.chunk --- simplex/epoch.go | 2 +- simplex/requestor.go | 17 +++++++++++++---- simplex/util.go | 33 +++++++++++++-------------------- simplex/util_test.go | 12 ++++++------ 4 files changed, 33 insertions(+), 31 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index 51f5d831..980535e4 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -30,7 +30,7 @@ const ( DefaultEmptyVoteRebroadcastTimeout = 5 * time.Second DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second EmptyVoteTimeoutID = "rebroadcast_empty_vote" - MaxRoundRequests = 10 + MaxRoundRequests = 10 // max number of rounds or sequences that fit in one replication request ) type EmptyVoteSet struct { diff --git a/simplex/requestor.go b/simplex/requestor.go index 0eaba228..c9a6da8f 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -122,6 +122,9 @@ 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 @@ -168,10 +171,15 @@ func (r *requestor) sendMoreReplicationRequests(observedSeqOrRound, currentSeqOr r.sendRequests(seqsOrRounds) } +// 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, numNodes, MaxRoundRequests) + batches := BatchSequences(seqsOrRounds, uint64(numNodes), MaxRoundRequests) for i, batch := range batches { index := (i + r.requestIterator) % numNodes @@ -180,9 +188,10 @@ func (r *requestor) sendRequests(seqsOrRounds []uint64) { 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. +// 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 { diff --git a/simplex/util.go b/simplex/util.go index 556aea06..9a093f7d 100644 --- a/simplex/util.go +++ b/simplex/util.go @@ -185,31 +185,24 @@ func (block *oneTimeVerifiedBlock) Verify(ctx context.Context) (common.VerifiedB return vb, err } -func BatchSequences(seqs []uint64, numNodes int, maxSize uint64) [][]uint64 { - if len(seqs) == 0 || numNodes <= 0 || maxSize == 0 { +// 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) numSeqs := uint64(len(seqs)) - share := numSeqs / uint64(numNodes) - remainder := numSeqs % uint64(numNodes) - - var batches [][]uint64 - start := uint64(0) - for i := 0; i < numNodes && start < numSeqs; i++ { - nodeShare := share - if uint64(i) < remainder { - nodeShare++ - } - for nodeShare > 0 { - n := min(nodeShare, maxSize) - batches = append(batches, seqs[start:start+n]) - start += n - nodeShare -= n - } - } - return batches + + seqsPerNode := (numSeqs + uint64(numNodes) - 1) / uint64(numNodes) + + batchSize := min(seqsPerNode, maxSize) + return slices.Collect(slices.Chunk(seqs, int(batchSize))) } type NotarizationTime struct { diff --git a/simplex/util_test.go b/simplex/util_test.go index 201d2f14..ad8373b9 100644 --- a/simplex/util_test.go +++ b/simplex/util_test.go @@ -307,8 +307,8 @@ func TestBatchSequences(t *testing.T) { maxSize: 10, expected: [][]uint64{ {0, 1, 2, 3}, - {4, 5, 6}, - {7, 8, 9}, + {4, 5, 6, 7}, + {8, 9}, }, }, { @@ -361,16 +361,16 @@ func TestBatchSequences(t *testing.T) { numNodes: 3, maxSize: 3, expected: [][]uint64{ - {0, 1, 2}, {3, 4}, - {5, 6, 7}, {8, 9}, - {10, 11, 12}, {13, 14}, + {0, 1, 2}, {3, 4, 5}, + {6, 7, 8}, {9, 10, 11}, + {12, 13, 14}, }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result := BatchSequences(tt.seqs, tt.numNodes, tt.maxSize) + result := BatchSequences(tt.seqs, uint64(tt.numNodes), tt.maxSize) require.Equal(t, tt.expected, result) }) } From a1009a23fe1e7ec62aff895d12cba90204aafec9 Mon Sep 17 00:00:00 2001 From: jadalsmail Date: Fri, 10 Jul 2026 14:55:48 -0400 Subject: [PATCH 8/8] fix TestReplicationResendSplitsRequests now tests resend reequests above the limit --- simplex/epoch.go | 2 +- simplex/replication_timeout_test.go | 94 ++++++++++++----------------- simplex/requestor.go | 6 +- 3 files changed, 45 insertions(+), 57 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index 7c7a6060..0483976c 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -30,7 +30,7 @@ const ( DefaultEmptyVoteRebroadcastTimeout = 5 * time.Second DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second EmptyVoteTimeoutID = "rebroadcast_empty_vote" - MaxRoundRequests = 10 // max number of rounds or sequences that fit in one replication request + maxItemCountPerRequest = 10 // max number of rounds or sequences that fit in one replication request ) type EmptyVoteSet struct { diff --git a/simplex/replication_timeout_test.go b/simplex/replication_timeout_test.go index 0c2f619c..80ca1f2d 100644 --- a/simplex/replication_timeout_test.go +++ b/simplex/replication_timeout_test.go @@ -678,69 +678,53 @@ func TestReplicationResendsFinalizedBlocksThatFailedVerification(t *testing.T) { 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() + // 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}} - sentMessages := make(chan *common.Message, 100) - conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[1], &recordingComm{ - Communication: testutil.NewNoopComm(nodes), - SentMessages: sentMessages, - }, bb) - conf.ReplicationEnabled = true + startSeq := 2 * uint64(simplex.DefaultMaxRoundWindow) - 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) + net := testutil.NewControlledNetwork(t, nodes) + storageData := createBlocks(t, nodes, startSeq) - // 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") + newNodeConfig := func(from common.NodeID) *testutil.TestNodeConfig { + comm := testutil.NewTestComm(from, net.BasicInMemoryNetwork, rejectReplicationRequests) + return &testutil.TestNodeConfig{ + InitialStorage: storageData, + Comm: comm, + ReplicationEnabled: true, } } - // no node responds, so the requests time out and are re-sent - e.AdvanceTime(e.StartTime.Add(2 * simplex.DefaultReplicationRequestTimeout)) + 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, + }) - // 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") + 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.Greater(t, resentRequests, 1, - "timed out requests should be re-sent split across the signers of the highest observed quorum") + 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) } diff --git a/simplex/requestor.go b/simplex/requestor.go index c9a6da8f..de005572 100644 --- a/simplex/requestor.go +++ b/simplex/requestor.go @@ -5,6 +5,7 @@ package simplex import ( "math" + "slices" "sync" "time" @@ -179,7 +180,7 @@ func (r *requestor) sendMoreReplicationRequests(observedSeqOrRound, currentSeqOr func (r *requestor) sendRequests(seqsOrRounds []uint64) { signers := r.highestObserved.signers numNodes := len(signers) - batches := BatchSequences(seqsOrRounds, uint64(numNodes), MaxRoundRequests) + batches := BatchSequences(seqsOrRounds, uint64(numNodes), maxItemCountPerRequest) for i, batch := range batches { index := (i + r.requestIterator) % numNodes @@ -208,6 +209,9 @@ func (r *requestor) sendRequestToNode(seqsOrRounds []uint64, node common.NodeID) return } + if !slices.IsSorted(toRequest) { + slices.Sort(toRequest) + } if last := toRequest[len(toRequest)-1]; last > r.highestRequested { r.highestRequested = last }