Skip to content

Protect against leaders that send several distinct blocks for the same round #570

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yacovm merged 6 commits into from
Sep 8, 2026
Merged

Protect against leaders that send several distinct blocks for the same round #570

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4dc6c0f
Protect against leaders that send several distinct blocks for the sam…
yacovm Sep 1, 2026
2847cbf
Address code review comments
yacovm Sep 3, 2026
794b28f
Address code review comments II
yacovm Sep 3, 2026
5f719ac
Merge branch 'main' into 541
yacovm Sep 3, 2026
d740c9f
Merge branch 'main' into 541
yacovm Sep 8, 2026
0b215a2
Merge branch 'main' into 541
yacovm Sep 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions simplex/epoch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1942,15 +1942,21 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod
return nil
}

// If we are already processing a block for this round, reject the block while it is being processed.
if msgForRound, exists := e.futureMessages[string(from)][md.Round]; exists && msgForRound.proposalBeingProcessed {
e.Logger.Debug("Got block for a round that is being processed", zap.Uint64("round", md.Round))
return nil
}

// Create a task that will verify the block in the future, after its predecessors have also been verified.
task := e.createBlockVerificationTask(e.oneTimeVerifier.Wrap(block), from, vote)

@samliok samliok Sep 3, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

where do we set this block as being processing? if the leader just sends via the handleBlockMessage() then don't we still run into this issue?

@yacovm yacovm Sep 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://github.com/ava-labs/Simplex/pull/570/changes/BASE..62de7c47ff4ae4cc6163c9a0b7474f276b3e8d56#diff-b6a2eb766dc190b568ec16bd7277c8633013e63dcd6e590db5c72dcc0b69d220R1948

@yacovm yacovm Sep 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

a few lines below

@yacovm yacovm Sep 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

	// mark in future messages while we are verifying the block
	msgForRound, exists := e.futureMessages[string(from)][md.Round]
	if !exists {
		msgsForRound := &messagesForRound{}
		msgsForRound.proposalBeingProcessed = true
		e.futureMessages[string(from)][md.Round] = msgsForRound
	} else {
		msgForRound.proposalBeingProcessed = true
	}


prevBlockDependency, missingRounds := e.blockDependencies(md)

if len(missingRounds) > 0 {
e.sendMissingRoundsRequest(from, missingRounds)
}

// Create a task that will verify the block in the future, after its predecessors have also been verified.
task := e.createBlockVerificationTask(e.oneTimeVerifier.Wrap(block), from, vote)

if err := e.blockVerificationScheduler.ScheduleTaskWithDependencies(task, md.Seq, prevBlockDependency, missingRounds); err != nil {
return nil
}
Expand Down Expand Up @@ -2175,6 +2181,8 @@ func (e *Epoch) createBlockVerificationTask(block common.Block, from common.Node
e.lock.Lock()
defer e.lock.Unlock()

e.deleteFutureProposal(from, md.Round)

if err != nil {
leader := LeaderForRound(e.validatorNodeIDs, md.Round)
e.Logger.Info("Triggering empty block agreement",
Expand All @@ -2186,8 +2194,6 @@ func (e *Epoch) createBlockVerificationTask(block common.Block, from common.Node
return md.Digest
}

e.deleteFutureProposal(from, md.Round)

if !e.storeProposal(verifiedBlock) {
e.Logger.Debug("Unable to store proposed block for the round", zap.Stringer("NodeID", from), zap.Uint64("round", md.Round))
return md.Digest
Expand Down
121 changes: 104 additions & 17 deletions simplex/epoch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -651,15 +651,18 @@ func TestEpochConsecutiveProposalsDoNotGetVerified(t *testing.T) {
name string
err error
expectedVerificationCount int
capacity int
}{
{
name: "valid block",
expectedVerificationCount: 1,
capacity: 10,
},
{
name: "invalid block",
err: fmt.Errorf("invalid block"),
expectedVerificationCount: DefaultProcessingBlocks,
expectedVerificationCount: 10,
capacity: 1,
},
} {
t.Run(test.name, func(t *testing.T) {
Expand All @@ -672,6 +675,15 @@ func TestEpochConsecutiveProposalsDoNotGetVerified(t *testing.T) {
require.NoError(t, err)
t.Cleanup(e.Stop)

semaphore := make(chan struct{}, test.capacity) // poor man's semaphore to limit the number of concurrent verifications

e.Logger.(*testutil.TestLogger).Intercept(func(entry zapcore.Entry) error {
if strings.Contains(entry.Message, "Block verification ended") {
<-semaphore // release a resource in the semaphore when a block verification ends
}
return nil
})

require.NoError(t, e.Start())

leader := nodes[0]
Expand All @@ -696,30 +708,105 @@ func TestEpochConsecutiveProposalsDoNotGetVerified(t *testing.T) {
vote, err := testutil.NewTestVote(block, leader)
require.NoError(t, err)

var wg sync.WaitGroup
wg.Add(DefaultProcessingBlocks)

for i := 0; i < DefaultProcessingBlocks; i++ {
go func() {
defer wg.Done()

err := e.HandleMessage(&Message{
BlockMessage: &BlockMessage{
Vote: *vote,
Block: block,
},
}, leader)
require.NoError(t, err)
}()
for i := 0; i < 10; i++ {
semaphore <- struct{}{} // acquire a resource in the semaphore before scheduling a block verification
err := e.HandleMessage(&Message{
BlockMessage: &BlockMessage{
Vote: *vote,
Block: block,
},
}, leader)
require.NoError(t, err)
}
wg.Wait()
scheduledWG.Wait()

require.Equal(t, uint32(test.expectedVerificationCount), timesVerified.Load())
})
}
}

// TestEpochLeaderEquivocationDoesNotFloodBlockVerification tests that
// a leader flooding distinct valid blocks for the same round must get at most
// one verified and must not overwhelm the bounded verification queue.
func TestEpochLeaderEquivocationDoesNotFloodBlockVerification(t *testing.T) {
bb := testutil.NewTestBlockBuilder()
nodes := []NodeID{{1}, {2}, {3}, {4}}

conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[1], testutil.NewNoopComm(nodes), bb)

// Count the scheduler's "queue full" warnings. Installed before NewEpoch (and
// thus before the scheduler goroutine is started) so there is no race on the logger.
var queueFullWarnings atomic.Uint32
logger := conf.Logger.(*testutil.TestLogger)
logger.Intercept(func(entry zapcore.Entry) error {
if strings.Contains(entry.Message, "Too many blocks being verified") {
queueFullWarnings.Add(1)
}
return nil
})

e, err := NewEpoch(conf)
require.NoError(t, err)

leader := LeaderForRound(nodes, 0)
require.NotEqual(t, conf.ID, leader) // the local node must not be the leader of the round

// Hold every verification until the flood is fully submitted, so exactly one
// proposal stays in-flight (pendingRounds only blocks an in-flight verification).
gate := make(chan struct{})
var releaseGate sync.Once
openGate := func() { releaseGate.Do(func() { close(gate) }) }
t.Cleanup(e.Stop)
t.Cleanup(openGate)

require.NoError(t, e.Start())

md := e.Metadata()
require.Equal(t, uint64(0), md.Round)

var timesVerified atomic.Uint32

// More distinct, valid proposals than the 500-slot queue can hold.
const floodSize = DefaultProcessingBlocks + 50
Comment thread
samliok marked this conversation as resolved.
digests := make(map[Digest]struct{}, floodSize)
for i := 0; i < floodSize; i++ {
// Distinct payload -> distinct digest, but identical, valid metadata.
block := testutil.NewTestBlock(md, emptyBlacklist)
block.Data = []byte(fmt.Sprintf("equivocated-proposal-%d", i))
block.ComputeDigest()
block.VerificationDelay = gate
block.OnVerify = func() { timesVerified.Add(1) }

d := block.BlockHeader().Digest
_, dup := digests[d]
require.False(t, dup, "the leader's equivocated blocks must be distinct")
digests[d] = struct{}{}

vote, err := testutil.NewTestVote(block, leader)
require.NoError(t, err)

require.NoError(t, e.HandleMessage(&Message{
BlockMessage: &BlockMessage{Vote: *vote, Block: block},
}, leader))
}

// Scheduling is synchronous within HandleMessage, so any queue-full warnings
// have already been emitted by the time the flood loop returns.
require.Zero(t, queueFullWarnings.Load(),
"leader equivocation flood must not overrun the shared verification queue")

// Let the scheduled verification(s) run to completion.
openGate()

// The leader's proposal is verified at least once...
require.Eventually(t, func() bool { return timesVerified.Load() >= 1 },
time.Second, 10*time.Millisecond, "the leader's proposal should be verified")
// ...but no more than one proposal per (leader, round) is verified.
require.Never(t, func() bool { return timesVerified.Load() > 1 },
500*time.Millisecond, 10*time.Millisecond,
"a leader must not get more than one proposal verified for the same round")
}

// TestEpochIncreasesRoundAfterFinalization ensures that the epochs round is incremented
// if we receive a finalization for the current round(even if it is not the next seq to commit)
func TestEpochIncreasesRoundAfterFinalization(t *testing.T) {
Expand Down
Loading