From 4dc6c0fd3108d6d65fb820209223e388ad0c60b3 Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Tue, 1 Sep 2026 19:52:51 +0200 Subject: [PATCH 1/3] Protect against leaders that send several distinct blocks for the same round This commit limits block verification at any round to a single block at a time when receiving the block as a block message. This is to prevent from a malicious leader from sending many distinct blocks and overwhelming the node. Signed-off-by: Yacov Manevich --- simplex/epoch.go | 33 ++++++++++-- simplex/epoch_test.go | 121 ++++++++++++++++++++++++++++++++++++------ 2 files changed, 134 insertions(+), 20 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index 7e3c95a6..cfd99ad2 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -106,6 +106,7 @@ type Epoch struct { validatorNodeIDs common.NodeIDs validators common.Nodes validatorsToPKs map[string][]byte + pendingRounds map[uint64]struct{} // block messages that are verified and may create a new round object soon rounds map[uint64]*Round emptyVotes map[uint64]*EmptyVoteSet oldestNotFinalizedNotarization NotarizationTime @@ -223,6 +224,7 @@ func (e *Epoch) init() error { e.timedOutRounds = make(map[uint16]uint64, len(e.validatorNodeIDs)) e.redeemedRounds = make(map[uint16]uint64, len(e.validatorNodeIDs)) e.rounds = make(map[uint64]*Round) + e.pendingRounds = make(map[uint64]struct{}) e.emptyVotes = make(map[uint64]*EmptyVoteSet) e.futureMessages = make(messagesFromNode, len(e.validatorNodeIDs)) e.replicationState = NewReplicationState(e.Logger, e.Comm, e.ID, e.MaxRoundWindow, e.ReplicationEnabled, e.StartTime, &e.lock, e.RandomSource) @@ -1908,6 +1910,16 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod return nil } + if e.isRoundPending(md.Round) { + e.Logger.Debug("Got block for a round that is pending", 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. + blockVerificationTask := e.createBlockVerificationTask(e.oneTimeVerifier.Wrap(block), from, vote) + // Mark the round as pending and wrap the task with a task that will cleanup the pending round after the block verification task is executed. + task := e.markPendingRoundAndCleanupAfter(md.Round, blockVerificationTask) + if !e.verifyProposalMetadataAndBlacklist(block) { e.Logger.Debug("Got invalid block in a BlockMessage") return nil @@ -1919,9 +1931,6 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod 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 } @@ -1949,6 +1958,24 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod return nil } +func (e *Epoch) isRoundPending(round uint64) bool { + _, exists := e.pendingRounds[round] + return exists +} + +// markPendingRoundAndCleanupAfter marks a round as pending and returns a function that will clean up the pending round after the task is executed. +func (e *Epoch) markPendingRoundAndCleanupAfter(round uint64, task common.Task) common.Task { + e.pendingRounds[round] = struct{}{} + return func() common.Digest { + defer func() { + e.lock.Lock() + defer e.lock.Unlock() + delete(e.pendingRounds, round) + }() + return task() + } +} + func (e *Epoch) sendMissingRoundsRequest(to common.NodeID, missingRounds []uint64) { e.Logger.Debug("Requesting missing empty notarizations for rounds", zap.Stringer("to", to), diff --git a/simplex/epoch_test.go b/simplex/epoch_test.go index 9a6fb0f2..b2ab869c 100644 --- a/simplex/epoch_test.go +++ b/simplex/epoch_test.go @@ -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) { @@ -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] @@ -696,23 +708,16 @@ 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()) @@ -720,6 +725,88 @@ func TestEpochConsecutiveProposalsDoNotGetVerified(t *testing.T) { } } +// 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 + 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) { From 2847cbfe801df4ab53f536e31b25a6a38d063783 Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Thu, 3 Sep 2026 23:07:12 +0200 Subject: [PATCH 2/3] Address code review comments Signed-off-by: Yacov Manevich --- simplex/epoch.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index cfd99ad2..b8add406 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -1910,6 +1910,11 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod return nil } + if !e.verifyProposalMetadataAndBlacklist(block) { + e.Logger.Debug("Got invalid block in a BlockMessage") + return nil + } + if e.isRoundPending(md.Round) { e.Logger.Debug("Got block for a round that is pending", zap.Uint64("round", md.Round)) return nil @@ -1920,11 +1925,6 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod // Mark the round as pending and wrap the task with a task that will cleanup the pending round after the block verification task is executed. task := e.markPendingRoundAndCleanupAfter(md.Round, blockVerificationTask) - if !e.verifyProposalMetadataAndBlacklist(block) { - e.Logger.Debug("Got invalid block in a BlockMessage") - return nil - } - prevBlockDependency, missingRounds := e.blockDependencies(md) if len(missingRounds) > 0 { From 794b28f046ce54ed983f94f25d25f54c810ceb2e Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Fri, 4 Sep 2026 00:38:14 +0200 Subject: [PATCH 3/3] Address code review comments II Signed-off-by: Yacov Manevich --- simplex/epoch.go | 33 ++++++--------------------------- 1 file changed, 6 insertions(+), 27 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index b8add406..00b7cf0a 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -106,7 +106,6 @@ type Epoch struct { validatorNodeIDs common.NodeIDs validators common.Nodes validatorsToPKs map[string][]byte - pendingRounds map[uint64]struct{} // block messages that are verified and may create a new round object soon rounds map[uint64]*Round emptyVotes map[uint64]*EmptyVoteSet oldestNotFinalizedNotarization NotarizationTime @@ -224,7 +223,6 @@ func (e *Epoch) init() error { e.timedOutRounds = make(map[uint16]uint64, len(e.validatorNodeIDs)) e.redeemedRounds = make(map[uint16]uint64, len(e.validatorNodeIDs)) e.rounds = make(map[uint64]*Round) - e.pendingRounds = make(map[uint64]struct{}) e.emptyVotes = make(map[uint64]*EmptyVoteSet) e.futureMessages = make(messagesFromNode, len(e.validatorNodeIDs)) e.replicationState = NewReplicationState(e.Logger, e.Comm, e.ID, e.MaxRoundWindow, e.ReplicationEnabled, e.StartTime, &e.lock, e.RandomSource) @@ -1915,15 +1913,14 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod return nil } - if e.isRoundPending(md.Round) { - e.Logger.Debug("Got block for a round that is pending", zap.Uint64("round", md.Round)) + // 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. - blockVerificationTask := e.createBlockVerificationTask(e.oneTimeVerifier.Wrap(block), from, vote) - // Mark the round as pending and wrap the task with a task that will cleanup the pending round after the block verification task is executed. - task := e.markPendingRoundAndCleanupAfter(md.Round, blockVerificationTask) + task := e.createBlockVerificationTask(e.oneTimeVerifier.Wrap(block), from, vote) prevBlockDependency, missingRounds := e.blockDependencies(md) @@ -1958,24 +1955,6 @@ func (e *Epoch) handleBlockMessage(message *common.BlockMessage, from common.Nod return nil } -func (e *Epoch) isRoundPending(round uint64) bool { - _, exists := e.pendingRounds[round] - return exists -} - -// markPendingRoundAndCleanupAfter marks a round as pending and returns a function that will clean up the pending round after the task is executed. -func (e *Epoch) markPendingRoundAndCleanupAfter(round uint64, task common.Task) common.Task { - e.pendingRounds[round] = struct{}{} - return func() common.Digest { - defer func() { - e.lock.Lock() - defer e.lock.Unlock() - delete(e.pendingRounds, round) - }() - return task() - } -} - func (e *Epoch) sendMissingRoundsRequest(to common.NodeID, missingRounds []uint64) { e.Logger.Debug("Requesting missing empty notarizations for rounds", zap.Stringer("to", to), @@ -2173,6 +2152,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", @@ -2184,8 +2165,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