diff --git a/common/msg.go b/common/msg.go index c6fc97d7..6679451f 100644 --- a/common/msg.go +++ b/common/msg.go @@ -25,6 +25,21 @@ type Message struct { BlockDigestRequest *BlockDigestRequest } +func (m *Message) IsReplicationMessage() bool { + switch { + case m.ReplicationResponse != nil: + return true + case m.ReplicationRequest != nil: + return true + case m.VerifiedReplicationResponse != nil: + return true + case m.BlockDigestRequest != nil: + return true + default: + return false + } +} + type EmptyVoteMetadata struct { Round uint64 Epoch uint64 diff --git a/common/msg_test.go b/common/msg_test.go index a2612a31..32567e68 100644 --- a/common/msg_test.go +++ b/common/msg_test.go @@ -1,3 +1,6 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + package common_test import ( @@ -9,6 +12,96 @@ import ( "github.com/stretchr/testify/require" ) +func TestIsReplicationMessage(t *testing.T) { + tests := []struct { + name string + msg common.Message + expected bool + }{ + { + name: "empty message", + msg: common.Message{}, + expected: false, + }, + { + name: "replication request", + msg: common.Message{ReplicationRequest: &common.ReplicationRequest{}}, + expected: true, + }, + { + name: "replication response", + msg: common.Message{ReplicationResponse: &common.ReplicationResponse{}}, + expected: true, + }, + { + name: "verified replication response", + msg: common.Message{VerifiedReplicationResponse: &common.VerifiedReplicationResponse{}}, + expected: true, + }, + { + name: "block message", + msg: common.Message{BlockMessage: &common.BlockMessage{}}, + expected: false, + }, + { + name: "verified block message", + msg: common.Message{VerifiedBlockMessage: &common.VerifiedBlockMessage{}}, + expected: false, + }, + { + name: "empty notarization", + msg: common.Message{EmptyNotarization: &common.EmptyNotarization{}}, + expected: false, + }, + { + name: "vote", + msg: common.Message{VoteMessage: &common.Vote{}}, + expected: false, + }, + { + name: "empty vote", + msg: common.Message{EmptyVoteMessage: &common.EmptyVote{}}, + expected: false, + }, + { + name: "notarization", + msg: common.Message{Notarization: &common.Notarization{}}, + expected: false, + }, + { + name: "finalize vote", + msg: common.Message{FinalizeVote: &common.FinalizeVote{}}, + expected: false, + }, + { + name: "finalization", + msg: common.Message{Finalization: &common.Finalization{}}, + expected: false, + }, + { + // A block digest request is a request but not a replication message. + name: "block digest request", + msg: common.Message{BlockDigestRequest: &common.BlockDigestRequest{}}, + expected: true, + }, + { + // When several fields are set, any replication field makes it a replication message. + name: "block message and replication request", + msg: common.Message{ + BlockMessage: &common.BlockMessage{}, + ReplicationRequest: &common.ReplicationRequest{}, + }, + expected: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.expected, test.msg.IsReplicationMessage()) + }) + } +} + func TestQuorumRoundMalformed(t *testing.T) { tests := []struct { name string diff --git a/simplex/epoch.go b/simplex/epoch.go index 13b10f70..d26b9ed3 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -82,6 +82,7 @@ type EpochConfig struct { type Epoch struct { EpochConfig // Runtime + epochSealed atomic.Bool signatureAggregator common.SignatureAggregator oneTimeVerifier *OneTimeVerifier buildBlockScheduler *common.BasicScheduler @@ -512,7 +513,7 @@ func (e *Epoch) broadcastReplicationSync() { LatestFinalizedSeq: latestFinalizedSeq, }, } - e.Comm.Broadcast(replicationRequest) + e.broadcast(replicationRequest) } // resumeFromWal resumes the epoch from the records of the write ahead log. @@ -522,7 +523,7 @@ func (e *Epoch) resumeFromWal(highestRoundRecord *walRound) error { // Handle the most relevant record based on priority: finalization > notarization > emptyNotarization > emptyVote > block if highestRoundRecord.finalization != nil { finalizationMsg := &common.Message{Finalization: highestRoundRecord.finalization} - e.Comm.Broadcast(finalizationMsg) + e.broadcast(finalizationMsg) e.Logger.Debug("Broadcast finalization", zap.Uint64("round", highestRoundRecord.finalization.Finalization.Round), @@ -534,7 +535,7 @@ func (e *Epoch) resumeFromWal(highestRoundRecord *walRound) error { if highestRoundRecord.notarization != nil { notarization := highestRoundRecord.notarization lastMessage := common.Message{Notarization: notarization} - e.Comm.Broadcast(&lastMessage) + e.broadcast(&lastMessage) if e.sequenceAlreadyIndexed(notarization.Vote.Seq) { e.Logger.Debug("Notarization already indexed, skipping restoration", zap.Uint64("Sequence", notarization.Vote.Seq)) @@ -546,7 +547,7 @@ func (e *Epoch) resumeFromWal(highestRoundRecord *walRound) error { if highestRoundRecord.emptyNotarization != nil { lastMessage := common.Message{EmptyNotarization: highestRoundRecord.emptyNotarization} - e.Comm.Broadcast(&lastMessage) + e.broadcast(&lastMessage) return e.startRound() } @@ -561,7 +562,7 @@ func (e *Epoch) resumeFromWal(highestRoundRecord *walRound) error { return fmt.Errorf("could not find my own vote for round %d", ev.Round) } lastMessage := common.Message{EmptyVoteMessage: emptyVote} - e.Comm.Broadcast(&lastMessage) + e.broadcast(&lastMessage) e.Logger.Info("Rebroadcasting empty vote from WAL", zap.Uint64("round", ev.Round)) e.addEmptyVoteRebroadcastTimeout() } @@ -591,7 +592,7 @@ func (e *Epoch) resumeFromWal(highestRoundRecord *walRound) error { }, } // broadcast only if we are the leader - e.Comm.Broadcast(proposal) + e.broadcast(proposal) err = e.handleVoteMessage(&vote, e.ID) if err != nil { return err @@ -736,6 +737,10 @@ func (e *Epoch) Stop() { e.replicationState.Close() } +func (e *Epoch) isEpochSealed() bool { + return e.epochSealed.Load() +} + func (e *Epoch) handleFinalizationMessage(message *common.Finalization, from common.NodeID) error { e.Logger.Verbo("Received finalization message", zap.Stringer("from", from), zap.Uint64("round", message.Finalization.Round), zap.Uint64("seq", message.Finalization.Seq)) @@ -841,7 +846,7 @@ func (e *Epoch) handleFinalizeVoteMessage(message *common.FinalizeVote, from com return nil } // send the finalization to the sender in case they missed it - e.Comm.Send(&common.Message{ + e.send(&common.Message{ Finalization: round.finalization, }, from) return nil @@ -947,7 +952,7 @@ func (e *Epoch) sendLatestFinalization(to common.NodeID) { Finalization: &e.lastBlock.Finalization, } e.Logger.Debug("Node appears behind, sending it the latest finalization", zap.Stringer("to", to), zap.Uint64("round", e.lastBlock.Finalization.Finalization.Round), zap.Uint64("sequence", e.lastBlock.Finalization.Finalization.Seq)) - e.Comm.Send(msg, to) + e.send(msg, to) } func (e *Epoch) sendHighestRound(to common.NodeID) { @@ -963,7 +968,7 @@ func (e *Epoch) sendHighestRound(to common.NodeID) { Notarization: latestQR.Notarization, } e.Logger.Debug("Node appears behind, sending it the highest round", zap.Stringer("to", to), zap.Uint64("round", latestQR.Notarization.Vote.Round)) - e.Comm.Send(msg, to) + e.send(msg, to) return } @@ -972,7 +977,7 @@ func (e *Epoch) sendHighestRound(to common.NodeID) { EmptyNotarization: latestQR.EmptyNotarization, } e.Logger.Debug("Node appears behind, sending it the highest empty notarized round", zap.Stringer("to", to), zap.Uint64("round", latestQR.EmptyNotarization.Vote.Round)) - e.Comm.Send(msg, to) + e.send(msg, to) return } } @@ -988,7 +993,7 @@ func (e *Epoch) maybeSendNotarizationOrFinalization(to common.NodeID, round uint EmptyNotarization: evs.emptyNotarization, } e.Logger.Debug("Node appears behind, sending it an empty notarization", zap.Stringer("to", to), zap.Uint64("round", round)) - e.Comm.Send(msg, to) + e.send(msg, to) } return @@ -998,7 +1003,7 @@ func (e *Epoch) maybeSendNotarizationOrFinalization(to common.NodeID, round uint msg := &common.Message{ Finalization: r.finalization, } - e.Comm.Send(msg, to) + e.send(msg, to) return } @@ -1007,7 +1012,7 @@ func (e *Epoch) maybeSendNotarizationOrFinalization(to common.NodeID, round uint msg := &common.Message{ Notarization: r.notarization, } - e.Comm.Send(msg, to) + e.send(msg, to) return } } @@ -1269,7 +1274,7 @@ func (e *Epoch) persistFinalization(finalization common.Finalization) error { } finalizationMsg := &common.Message{Finalization: &finalization} - e.Comm.Broadcast(finalizationMsg) + e.broadcast(finalizationMsg) e.Logger.Debug("Broadcast finalization", zap.Uint64("round", finalization.Finalization.Round), @@ -1318,12 +1323,32 @@ func (e *Epoch) rebroadcastPastFinalizeVotes() error { finalizeVoteMessage = msg } e.Logger.Debug("Rebroadcasting finalize vote", zap.Uint64("round", r), zap.Uint64("seq", finalizeVoteMessage.FinalizeVote.Finalization.Seq)) - e.Comm.Broadcast(finalizeVoteMessage) + e.broadcast(finalizeVoteMessage) } return nil } +func (e *Epoch) broadcast(msg *common.Message) { + // If the epoch is sealed, we only allow messages related to replication to be broadcasted. + // This is to prevent advancing the entire protocol state, but allowing replicating past protocol state. + if e.isEpochSealed() && !msg.IsReplicationMessage() { + e.Logger.Debug("Epoch is sealed, aborting message broadcast") + return + } + e.Comm.Broadcast(msg) +} + +func (e *Epoch) send(msg *common.Message, to common.NodeID) { + // If the epoch is sealed, we only allow messages related to replication to be sent. + // This is to prevent advancing the entire protocol state, but allowing replicating past protocol state. + if e.isEpochSealed() && !msg.IsReplicationMessage() { + e.Logger.Debug("Epoch is sealed, aborting message send", zap.Stringer("to", to)) + return + } + e.Comm.Send(msg, to) +} + func (e *Epoch) maxRoundInRoundsMap() uint64 { maxRound := uint64(0) for r := range e.rounds { @@ -1400,6 +1425,20 @@ func (e *Epoch) indexFinalization(block common.VerifiedBlock, finalization commo Finalization: finalization, } + // If this is a sealing block, and is not the zero block, then we have sealed the epoch and should stop processing messages. + // In case it's the zero block, we do not not seal the epoch. + if block.SealingBlockInfo() != nil && block.SealingBlockInfo().PrevSealingBlockHash != [32]byte{} { + e.Logger.Info("Committed a sealing block, epoch is sealed", + zap.Uint64("round", finalization.Finalization.Round), + zap.Uint64("sequence", finalization.Finalization.Seq), + zap.Stringer("digest", finalization.Finalization.BlockHeader.Digest)) + + finalizationMsg := &common.Message{Finalization: &finalization} + e.broadcast(finalizationMsg) + + e.epochSealed.Store(true) + } + // We have committed because we have collected a finalization. // However, we may have not witnessed a notarization. // Regardless of that, we can safely progress to the round succeeding the finalization. @@ -1473,7 +1512,7 @@ func (e *Epoch) persistEmptyNotarization(emptyVotes *EmptyVoteSet, shouldBroadca emptyVotes.persisted = true if shouldBroadcast { notarizationMessage := &common.Message{EmptyNotarization: emptyNotarization} - e.Comm.Broadcast(notarizationMessage) + e.broadcast(notarizationMessage) e.Logger.Debug("Broadcast empty notarization", zap.Uint64("round", emptyNotarization.Vote.Round)) } @@ -1600,7 +1639,7 @@ func (e *Epoch) persistAndBroadcastNotarization(notarization common.Notarization } notarizationMessage := &common.Message{Notarization: ¬arization} - e.Comm.Broadcast(notarizationMessage) + e.broadcast(notarizationMessage) e.Logger.Debug("Broadcast notarization", zap.Uint64("round", notarization.Vote.Round), @@ -1719,7 +1758,7 @@ func (e *Epoch) handleNotarizationMessage(message *common.Notarization, from com Seq: vote.Seq, }, } - e.Comm.Send(blockDigestRequest, from) + e.send(blockDigestRequest, from) return nil } @@ -1850,7 +1889,7 @@ func (e *Epoch) sendMissingRoundsRequest(to common.NodeID, missingRounds []uint6 }, } - e.Comm.Send(request, to) + e.send(request, to) } // blockDependencies returns the dependencies bh has before it can be verified. @@ -2016,6 +2055,10 @@ func (e *Epoch) createBlockVerificationTask(block common.Block, from common.Node e.Logger.Debug("Block verification ended", zap.Uint64("round", md.Round), zap.Duration("elapsed", elapsed)) }() + if e.isEpochSealed() { + e.Logger.Debug("Aborting block verification because the epoch is sealed", zap.Uint64("seq", md.Seq)) + return md.Digest + } verifiedBlock, err := block.Verify(context.Background()) e.lock.Lock() @@ -2104,6 +2147,11 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz e.Logger.Debug("Block verification ended", zap.Uint64("round", md.Round), zap.Duration("elapsed", elapsed)) }() + if e.isEpochSealed() { + e.Logger.Debug("Aborting block verification because the epoch is sealed", zap.Uint64("seq", md.Seq)) + return md.Digest + } + verifiedBlock, err := block.Verify(context.Background()) if err != nil { e.Logger.Debug("Failed verifying block", zap.Error(err)) @@ -2172,6 +2220,11 @@ func (e *Epoch) createNotarizedBlockVerificationTask(block common.Block, notariz e.Logger.Debug("Block verification ended", zap.Uint64("round", md.Round), zap.Duration("elapsed", elapsed)) }() + if e.isEpochSealed() { + e.Logger.Debug("Aborting block verification because the epoch is sealed", zap.Uint64("seq", md.Seq)) + return md.Digest + } + verifiedBlock, err := block.Verify(context.Background()) if err != nil { e.Logger.Debug("Failed verifying block", zap.Error(err)) @@ -2454,6 +2507,14 @@ func (e *Epoch) createBlockBuildingTask(metadata common.ProtocolMetadata, blackl cancel := e.blockBuilderCancelFunc return func() common.Digest { + e.lock.Lock() + if e.isEpochSealed() { + e.lock.Unlock() + e.Logger.Debug("Aborting block building because the epoch is sealed") + return common.Digest{} + } + e.lock.Unlock() + block, ok := e.BlockBuilder.BuildBlock(context, metadata, blacklist) e.lock.Lock() @@ -2512,7 +2573,7 @@ func (e *Epoch) proposeBlock(block common.VerifiedBlock) error { }, } - e.Comm.Broadcast(proposal) + e.broadcast(proposal) e.Logger.Debug("Proposal broadcast", zap.Uint64("round", md.Round), zap.Int("size", len(rawBlock)), @@ -2595,7 +2656,7 @@ func (e *Epoch) triggerEmptyBlockNotarization(round uint64) { // Add our own empty vote to the set emptyVotes.votes[string(e.ID)] = &signedEV - e.Comm.Broadcast(&common.Message{EmptyVoteMessage: &signedEV}) + e.broadcast(&common.Message{EmptyVoteMessage: &signedEV}) e.addEmptyVoteRebroadcastTimeout() @@ -2623,7 +2684,7 @@ func (e *Epoch) emptyVoteTimeoutTaskRunner(_ []string) { } e.Logger.Debug("Rebroadcasting empty vote because round has not advanced", zap.Uint64("round", ourVote.Vote.Round)) - e.Comm.Broadcast(&common.Message{EmptyVoteMessage: ourVote}) + e.broadcast(&common.Message{EmptyVoteMessage: ourVote}) } func (e *Epoch) addEmptyVoteRebroadcastTimeout() { @@ -2795,7 +2856,7 @@ func (e *Epoch) doProposed(block common.VerifiedBlock) error { zap.Uint64("round", md.Round), zap.Stringer("digest", md.Digest)) - e.Comm.Broadcast(voteMsg) + e.broadcast(voteMsg) // Send yourself a vote message return e.handleVoteMessage(&vote, e.ID) } @@ -2870,7 +2931,7 @@ func (e *Epoch) doNotarized(r uint64) error { if err != nil { return err } - e.Comm.Broadcast(finalizeVoteMsg) + e.broadcast(finalizeVoteMsg) e.Logger.Debug("Broadcasting finalize vote", zap.Uint64("round", md.Round), @@ -3072,7 +3133,7 @@ func (e *Epoch) handleReplicationRequest(req *common.ReplicationRequest, from co e.Logger.Debug("Sending response back to node", zap.Stringer("to", from), zap.Int("num rounds", len(data))) msg := &common.Message{VerifiedReplicationResponse: response} - e.Comm.Send(msg, from) + e.send(msg, from) return nil } @@ -3202,7 +3263,7 @@ func (e *Epoch) handleBlockDigestRequest(req *common.BlockDigestRequest, from co } msg := &common.Message{VerifiedReplicationResponse: response} - e.Comm.Send(msg, from) + e.send(msg, from) return nil } diff --git a/simplex/epoch_sealed_test.go b/simplex/epoch_sealed_test.go new file mode 100644 index 00000000..a348dbb1 --- /dev/null +++ b/simplex/epoch_sealed_test.go @@ -0,0 +1,334 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex_test + +import ( + "strings" + "sync/atomic" + "testing" + "time" + + . "github.com/ava-labs/simplex/common" + . "github.com/ava-labs/simplex/simplex" + "github.com/ava-labs/simplex/testutil" + + "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" +) + +const ( + // sealingTestBlocks is the number of blocks (rounds/seqs 0..8) proposed by other nodes. + // It is also the round the epoch node itself leads (round 9). + sealingTestBlocks = uint64(9) + // sealingTestSealingSeq is the sequence of the (potential) sealing block - the 5th block. + sealingTestSealingSeq = uint64(4) +) + +// sealingBlockTestEnv is the fixture produced by setupSealingBlockTest. It holds a started +// epoch and the block chain, so the test body can drive the epoch with messages and then +// assert on the result. +type sealingBlockTestEnv struct { + e *Epoch + comm *recordingComm + storage *testutil.InMemStorage + nodes []NodeID + quorum int + sigAggr SignatureAggregator + blocks []*testutil.TestBlock + verified map[uint64]*atomic.Bool + sealed *atomic.Bool + buildAborted *atomic.Bool +} + +// allVerified reports whether every block in the chain has been verified. +func (env *sealingBlockTestEnv) allVerified() bool { + for seq := uint64(0); seq < sealingTestBlocks; seq++ { + if !env.verified[seq].Load() { + return false + } + } + return true +} + +// verifiedAfterSealingBlock reports whether any block after the sealing block has been verified. +func (env *sealingBlockTestEnv) verifiedAfterSealingBlock() bool { + for seq := sealingTestSealingSeq + 1; seq < sealingTestBlocks; seq++ { + if env.verified[seq].Load() { + return true + } + } + return false +} + +// proposedLeaderRound reports whether the epoch has broadcast a proposal for the round it leads. +func (env *sealingBlockTestEnv) proposedLeaderRound() bool { + return leaderProposedBlock(env.comm, sealingTestBlocks) +} + +func TestEpochStopsAtSealingBlock(t *testing.T) { + // Verifies that a sealing block stops the epoch from + // executing (verifying) any block that comes after it. + // + // The epoch instance is the last node in the leadership order of a 10-node membership, + // so it never proposes any of the first 9 blocks (rounds 0-8) - each is proposed by a + // different node - but it is the leader of the last round (round 9). Each case delivers + // all 9 blocks as block messages and then feeds finalizations for the first 5 sequences + // (0-4) and notarizations for the remaining 4 (5-8), driving the epoch to the round it + // leads, and then asserts: + // - Without a sealing block: all 9 blocks are verified, showing the messages do schedule + // every block for execution, and the epoch then proposes a block for the round it is the leader of. + // - With the 5th block (seq 4) being a sealing block: only the blocks up to and including + // the sealing block are verified and committed; the rest are not, because committing the + // sealing block seals the epoch and stops it from verifying any later block. + + for _, tc := range []struct { + name string + sealing bool + verify func(t *testing.T, env *sealingBlockTestEnv) + }{ + { + name: "without sealing block", + sealing: false, + verify: func(t *testing.T, env *sealingBlockTestEnv) { + // Without a sealing block, every block should eventually be verified. + require.Eventually(t, env.allVerified, 5*time.Second, 10*time.Millisecond, + "all blocks should be verified when there is no sealing block") + require.False(t, env.sealed.Load(), "epoch should not seal when there is no sealing block") + + // Having advanced through every preceding round, the epoch reaches the round + // it is the leader of and, because the epoch is not sealed, proposes a block for that round. + require.Eventually(t, env.proposedLeaderRound, 5*time.Second, 10*time.Millisecond, + "the epoch node should propose a block for the round it leads when the epoch is not sealed") + }, + }, + { + name: "with sealing block", + sealing: true, + verify: func(t *testing.T, env *sealingBlockTestEnv) { + // The blocks after the sealing block must never be verified. + require.Never(t, env.verifiedAfterSealingBlock, 500*time.Millisecond, 50*time.Millisecond, + "no block after the sealing block should be verified") + + require.True(t, env.sealed.Load(), "epoch should have been sealed after committing the sealing block") + require.Equal(t, sealingTestSealingSeq+1, env.storage.NumBlocks(), + "only the blocks up to and including the sealing block should be committed") + for seq := uint64(0); seq <= sealingTestSealingSeq; seq++ { + require.True(t, env.verified[seq].Load(), "block %d (up to the sealing block) should have been verified", seq) + } + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + env := setupSealingBlockTest(t, sealingTestBlocks, tc.sealing) + + // Deliver every block as a block message from the block's leader. + for i := uint64(0); i < sealingTestBlocks; i++ { + leader := LeaderForRound(env.nodes, i) + vote, err := testutil.NewTestVote(env.blocks[i], leader) + require.NoError(t, err) + err = env.e.HandleMessage(&Message{ + BlockMessage: &BlockMessage{ + Block: env.blocks[i], + Vote: *vote, + }, + }, leader) + require.NoError(t, err) + } + + // Feed finalizations for the first 5 sequences (0-4) up to and including the sealing block (seq 4). + for i := uint64(0); i <= sealingTestSealingSeq; i++ { + finalization, _ := testutil.NewFinalizationRecord(t, env.sigAggr, env.blocks[i], env.nodes[:env.quorum]) + require.NoError(t, env.e.HandleMessage(&Message{Finalization: &finalization}, env.nodes[0])) + env.storage.WaitForBlockCommit(i) + } + + // Notarize the remaining 4 sequences (5-8). + for i := sealingTestSealingSeq + 1; i < sealingTestBlocks; i++ { + notarization, err := testutil.NewNotarization(env.e.Logger, env.sigAggr, env.blocks[i], env.nodes[:env.quorum]) + require.NoError(t, err) + testutil.InjectTestNotarization(t, env.e, notarization, env.nodes[0]) + } + + tc.verify(t, env) + }) + } +} + +func TestEpochBuildsInRoundAfterSealingBlock(t *testing.T) { + // Tests that we don't propose a block after a sealing block. + // + // The epoch node is chosen to be the leader of round sealingSeq+1. Each case delivers and + // finalizes blocks 0..sealingSeq (all led by other nodes); committing the last block + // progresses the epoch to the next round, which the epoch node leads. Each case then asserts: + // - Without a sealing block: the epoch is not sealed, so it builds and proposes a block for + // the round it leads. + // - With the last block (seq 4) being a sealing block: committing it seals the epoch, so + // when the epoch reaches the round it leads it aborts block building and proposes nothing. + + const roundAfterSealing = sealingTestSealingSeq + 1 + + for _, tc := range []struct { + name string + sealing bool + verify func(t *testing.T, env *sealingBlockTestEnv) + }{ + { + name: "without sealing block", + sealing: false, + verify: func(t *testing.T, env *sealingBlockTestEnv) { + // Not sealed, so the epoch node builds and proposes a block for the round it leads. + require.Eventually(t, func() bool { return leaderProposedBlock(env.comm, roundAfterSealing) }, 5*time.Second, 10*time.Millisecond, + "the epoch node should propose a block for the round it leads right after the block") + require.False(t, env.sealed.Load(), "epoch should not seal when there is no sealing block") + }, + }, + { + name: "with sealing block", + sealing: true, + verify: func(t *testing.T, env *sealingBlockTestEnv) { + // The epoch reaches the round it leads and attempts to build, but aborts because + // it is sealed - it does not build a block after the sealing block. Waiting on the + // abort also synchronizes past the seal, which happens right before it. + require.Eventually(t, env.buildAborted.Load, 5*time.Second, 10*time.Millisecond, + "a sealed epoch should abort block building in the round it leads") + require.True(t, env.sealed.Load(), "epoch should have been sealed after committing the sealing block") + + // It must never propose a block for that round either. + require.Never(t, func() bool { return leaderProposedBlock(env.comm, roundAfterSealing) }, 500*time.Millisecond, 50*time.Millisecond, + "a sealed epoch must not propose a block in the round it leads") + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + env := setupSealingBlockTest(t, roundAfterSealing, tc.sealing) + + // Deliver blocks 0..sealingSeq as block messages from their leaders (none is the epoch node). + for i := uint64(0); i <= sealingTestSealingSeq; i++ { + leader := LeaderForRound(env.nodes, i) + vote, err := testutil.NewTestVote(env.blocks[i], leader) + require.NoError(t, err) + require.NoError(t, env.e.HandleMessage(&Message{ + BlockMessage: &BlockMessage{ + Block: env.blocks[i], + Vote: *vote, + }, + }, leader)) + } + + // Finalize 0..sealingSeq. Committing the last block progresses the epoch to the + // next round, where the epoch node is the leader. + for i := uint64(0); i <= sealingTestSealingSeq; i++ { + finalization, _ := testutil.NewFinalizationRecord(t, env.sigAggr, env.blocks[i], env.nodes[:env.quorum]) + require.NoError(t, env.e.HandleMessage(&Message{Finalization: &finalization}, env.nodes[0])) + env.storage.WaitForBlockCommit(i) + } + + tc.verify(t, env) + }) + } +} + +// setupSealingBlockTest builds and starts a 10-node epoch whose node is the leader of +// epochLeaderRound, and constructs the chain of blocks the test will feed it. When +// markSealing is true, the seq-4 block is marked as a sealing block. It returns a fixture +// the test body uses to drive the epoch and assert on the result. +func setupSealingBlockTest(t *testing.T, epochLeaderRound uint64, markSealing bool) *sealingBlockTestEnv { + nodes := make([]NodeID, 10) + for i := range nodes { + nodes[i] = NodeID{byte(i + 1)} + } + + // Leadership is round-robin over the (distinct) nodes, so the epoch node leads + // epochLeaderRound and no other round in the [0, len(nodes)) range. + epochNode := LeaderForRound(nodes, epochLeaderRound) + + quorum := Quorum(len(nodes)) + blacklist := Blacklist{NodeCount: uint16(len(nodes)), SuspectedNodes: SuspectedNodes{}, Updates: []BlacklistUpdate{}} + + bb := testutil.NewTestBlockBuilder() + comm := &recordingComm{ + Communication: testutil.NewNoopComm(NodeIDs(nodes)), + SentMessages: make(chan *Message, 1000), + BroadcastMessages: make(chan *Message, 1000), + } + conf, _, storage := testutil.DefaultTestNodeEpochConfig(t, epochNode, comm, bb) + conf.ReplicationEnabled = true + + // Intercept logs so we can observe when the sealing block seals the epoch, and when a + // sealed epoch aborts building a block in a round it leads. + sealed := &atomic.Bool{} + buildAborted := &atomic.Bool{} + l := conf.Logger.(*testutil.TestLogger) + l.Intercept(func(entry zapcore.Entry) error { + if strings.Contains(entry.Message, "Committed a sealing block, epoch is sealed") { + sealed.Store(true) + } + if strings.Contains(entry.Message, "Aborting block building because the epoch is sealed") { + buildAborted.Store(true) + } + return nil + }) + + e, err := NewEpoch(conf) + require.NoError(t, err) + e.ReplicationEnabled = true + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + // Build a chain of 9 blocks (rounds/seqs 0..8), recording which blocks are verified. + verified := make(map[uint64]*atomic.Bool) + for i := uint64(0); i < sealingTestBlocks; i++ { + verified[i] = &atomic.Bool{} + } + + blocks := make([]*testutil.TestBlock, sealingTestBlocks) + var prev Digest + for i := uint64(0); i < sealingTestBlocks; i++ { + md := ProtocolMetadata{Round: i, Seq: i, Prev: prev} + block := testutil.NewTestBlock(md, blacklist) + seq := i + block.OnVerify = func() { + verified[seq].Store(true) + } + blocks[i] = block + prev = block.BlockHeader().Digest + } + + // If we are sealing the epoch, mark the sealing block with a validator set so the epoch knows it is a sealing block. + if markSealing { + blocks[sealingTestSealingSeq].SealingInfo = &SealingBlockInfo{ + ValidatorSet: NodeIDs(nodes).EqualWeightedNodes(), + PrevSealingBlockHash: blocks[0].Digest, + } + } + + sigAggr := e.SignatureAggregatorCreator(conf.Comm.Validators()) + + return &sealingBlockTestEnv{ + e: e, + comm: comm, + storage: storage, + nodes: nodes, + quorum: quorum, + sigAggr: sigAggr, + blocks: blocks, + verified: verified, + sealed: sealed, + buildAborted: buildAborted, + } +} + +// leaderProposedBlock reports whether the epoch has broadcast a proposal for the given round. +func leaderProposedBlock(comm *recordingComm, round uint64) bool { + for { + select { + case msg := <-comm.BroadcastMessages: + if msg.VerifiedBlockMessage != nil && msg.VerifiedBlockMessage.VerifiedBlock.BlockHeader().Round == round { + return true + } + default: + return false + } + } +} diff --git a/testutil/block.go b/testutil/block.go index 6d6a2d9e..7d067870 100644 --- a/testutil/block.go +++ b/testutil/block.go @@ -21,6 +21,8 @@ type TestBlock struct { OnVerify func() VerificationDelay chan struct{} VerificationError error + // SealingInfo, when non-nil, is returned by SealingBlockInfo() to mark this block as a sealing block. + SealingInfo *common.SealingBlockInfo } func NewTestBlock(metadata common.ProtocolMetadata, blacklist common.Blacklist) *TestBlock { @@ -59,7 +61,7 @@ func (tb *TestBlock) Verify(context.Context) (common.VerifiedBlock, error) { } func (tb *TestBlock) SealingBlockInfo() *common.SealingBlockInfo { - return nil + return tb.SealingInfo } func (tb *TestBlock) ComputeDigest() {