Merge pull request #134 from nspcc-dev/extend-processblock

*: take care of anti-MEV part

CHANGELOG.md

@@ -9,16 +9,21 @@ New features:
 Behaviour changes:
  * adjust behaviour of ProcessPreBlock callback (#129)
  * (*DBFT).Header() and (*DBFT).PreHeader() are moved to (*Context) receiver (#133)
+ * support error handling for ProcessBlock callback if anti-MEV extension is enabled
+   (#134)
 
 Improvements:
  * minimum required Go version is 1.22 (#122, #126)
+ * log Commit signature verification error (#134)
+ * add Commit message verification callback (#134)
 
 Bugs fixed:
  * context-bound PreBlock and PreHeader are not reset properly (#127)   
  * PreHeader is constructed instead of PreBlock to create PreCommit message (#128)
  * enable anti-MEV extension with respect to the current block index (#132)
  * (*Context).PreBlock() method returns PreHeader instead of PreBlock (#133)
  * WatchOnly node may send RecoveryMessage on RecoveryRequest (#135)
+ * invalid PreCommit message is not removed from cache (#134)
 
 ## [0.3.0] (01 August 2024)
 

check.go

@@ -112,8 +112,6 @@ func (d *DBFT[H]) checkCommit() {
 		return
 	}
 
-	d.lastBlockIndex = d.BlockIndex
-	d.lastBlockTime = d.Timer.Now()
 	d.block = d.CreateBlock()
 	hash := d.block.Hash()
 
@@ -124,8 +122,20 @@ func (d *DBFT[H]) checkCommit() {
 		zap.Stringer("merkle", d.block.MerkleRoot()),
 		zap.Stringer("prev", d.block.PrevHash()))
 
+	err := d.ProcessBlock(d.block)
+	if err != nil {
+		if d.isAntiMEVExtensionEnabled() {
+			d.Logger.Info("can't process Block, waiting for more Commits to be collected",
+				zap.Error(err),
+				zap.Int("count", count))
+			return
+		}
+		d.Logger.Fatal("block processing failed", zap.Error(err))
+	}
+
+	d.lastBlockIndex = d.BlockIndex
+	d.lastBlockTime = d.Timer.Now()
 	d.blockProcessed = true
-	d.ProcessBlock(d.block)
 
 	// Do not initialize consensus process immediately. It's the caller's duty to
 	// start the new block acceptance process and call Reset at the

config.go

@@ -51,7 +51,7 @@ type Config[H Hash] struct {
 	// ProcessBlock is called every time new preBlock is accepted.
 	ProcessPreBlock func(b PreBlock[H]) error
 	// ProcessBlock is called every time new block is accepted.
-	ProcessBlock func(b Block[H])
+	ProcessBlock func(b Block[H]) error
 	// GetBlock should return block with hash.
 	GetBlock func(h H) Block[H]
 	// WatchOnly tells if a node should only watch.
@@ -89,6 +89,10 @@ type Config[H Hash] struct {
 	// Note that PreBlock-dependent PreCommit verification should be performed inside PreBlock.Verify
 	// callback.
 	VerifyPreCommit func(p ConsensusPayload[H]) error
+	// VerifyCommit performs external Commit verification and returns nil if it's successful.
+	// Note that Block-dependent Commit verification should be performed inside Block.Verify
+	// callback.
+	VerifyCommit func(p ConsensusPayload[H]) error
 }
 
 const defaultSecondsPerBlock = time.Second * 15
@@ -108,7 +112,7 @@ func defaultConfig[H Hash]() *Config[H] {
 		GetVerified:        func() []Transaction[H] { return make([]Transaction[H], 0) },
 		VerifyBlock:        func(Block[H]) bool { return true },
 		Broadcast:          func(ConsensusPayload[H]) {},
-		ProcessBlock:       func(Block[H]) {},
+		ProcessBlock:       func(Block[H]) error { return nil },
 		GetBlock:           func(H) Block[H] { return nil },
 		WatchOnly:          func() bool { return false },
 		CurrentHeight:      nil,
@@ -117,6 +121,7 @@ func defaultConfig[H Hash]() *Config[H] {
 
 		VerifyPrepareRequest:  func(ConsensusPayload[H]) error { return nil },
 		VerifyPrepareResponse: func(ConsensusPayload[H]) error { return nil },
+		VerifyCommit:          func(ConsensusPayload[H]) error { return nil },
 
 		AntiMEVExtensionEnablingHeight: -1,
 		VerifyPreBlock:                 func(PreBlock[H]) bool { return true },
@@ -275,8 +280,9 @@ func WithBroadcast[H Hash](f func(m ConsensusPayload[H])) func(config *Config[H]
 	}
 }
 
-// WithProcessBlock sets ProcessBlock.
-func WithProcessBlock[H Hash](f func(b Block[H])) func(config *Config[H]) {
+// WithProcessBlock sets ProcessBlock callback. Note that for anti-MEV extension
+// disabled non-nil error return is a no-op.
+func WithProcessBlock[H Hash](f func(b Block[H]) error) func(config *Config[H]) {
 	return func(cfg *Config[H]) {
 		cfg.ProcessBlock = f
 	}
@@ -400,3 +406,10 @@ func WithVerifyPreCommit[H Hash](f func(preCommit ConsensusPayload[H]) error) fu
 		cfg.VerifyPreCommit = f
 	}
 }
+
+// WithVerifyCommit sets VerifyCommit.
+func WithVerifyCommit[H Hash](f func(commit ConsensusPayload[H]) error) func(config *Config[H]) {
+	return func(cfg *Config[H]) {
+		cfg.VerifyCommit = f
+	}
+}

dbft.go

@@ -537,6 +537,7 @@ func (d *DBFT[H]) onPreCommit(msg ConsensusPayload[H]) {
 	d.PreCommitPayloads[msg.ValidatorIndex()] = msg
 	if d.ViewNumber == msg.ViewNumber() {
 		if err := d.VerifyPreCommit(msg); err != nil {
+			d.PreCommitPayloads[msg.ValidatorIndex()] = nil
 			d.Logger.Warn("invalid PreCommit", zap.Uint16("from", msg.ValidatorIndex()), zap.String("error", err.Error()))
 			return
 		}
@@ -585,17 +586,24 @@ func (d *DBFT[H]) onCommit(msg ConsensusPayload[H]) {
 	}
 	d.CommitPayloads[msg.ValidatorIndex()] = msg
 	if d.ViewNumber == msg.ViewNumber() {
+		if err := d.VerifyCommit(msg); err != nil {
+			d.CommitPayloads[msg.ValidatorIndex()] = nil
+			d.Logger.Warn("invalid Commit", zap.Uint16("from", msg.ValidatorIndex()), zap.String("error", err.Error()))
+			return
+		}
+
 		d.Logger.Info("received Commit", zap.Uint("validator", uint(msg.ValidatorIndex())))
 		d.extendTimer(4)
 		header := d.MakeHeader()
 		if header != nil {
 			pub := d.Validators[msg.ValidatorIndex()]
-			if header.Verify(pub, msg.GetCommit().Signature()) == nil {
+			if err := header.Verify(pub, msg.GetCommit().Signature()); err == nil {
 				d.checkCommit()
 			} else {
 				d.CommitPayloads[msg.ValidatorIndex()] = nil
 				d.Logger.Warn("invalid commit signature",
 					zap.Uint("validator", uint(msg.ValidatorIndex())),
+					zap.Error(err),
 				)
 			}
 		}

dbft_test.go

@@ -1131,7 +1131,7 @@ func (s *testState) getOptions() []func(*dbft.Config[crypto.Uint256]) {
 		}),
 		dbft.WithBroadcast[crypto.Uint256](func(p Payload) { s.ch = append(s.ch, p) }),
 		dbft.WithGetTx[crypto.Uint256](s.pool.Get),
-		dbft.WithProcessBlock[crypto.Uint256](func(b dbft.Block[crypto.Uint256]) { s.blocks = append(s.blocks, b) }),
+		dbft.WithProcessBlock[crypto.Uint256](func(b dbft.Block[crypto.Uint256]) error { s.blocks = append(s.blocks, b); return nil }),
 		dbft.WithWatchOnly[crypto.Uint256](func() bool { return false }),
 		dbft.WithGetBlock[crypto.Uint256](func(crypto.Uint256) dbft.Block[crypto.Uint256] { return nil }),
 		dbft.WithTimer[crypto.Uint256](timer.New()),
@@ -1150,6 +1150,7 @@ func (s *testState) getOptions() []func(*dbft.Config[crypto.Uint256]) {
 		dbft.WithNewRecoveryMessage[crypto.Uint256](func() dbft.RecoveryMessage[crypto.Uint256] {
 			return consensus.NewRecoveryMessage(nil)
 		}),
+		dbft.WithVerifyCommit[crypto.Uint256](func(p dbft.ConsensusPayload[crypto.Uint256]) error { return nil }),
 	}
 
 	verify := s.verify

internal/consensus/consensus.go

@@ -13,7 +13,7 @@ func New(logger *zap.Logger, key dbft.PrivateKey, pub dbft.PublicKey,
 	getTx func(uint256 crypto.Uint256) dbft.Transaction[crypto.Uint256],
 	getVerified func() []dbft.Transaction[crypto.Uint256],
 	broadcast func(dbft.ConsensusPayload[crypto.Uint256]),
-	processBlock func(dbft.Block[crypto.Uint256]),
+	processBlock func(dbft.Block[crypto.Uint256]) error,
 	currentHeight func() uint32,
 	currentBlockHash func() crypto.Uint256,
 	getValidators func(...dbft.Transaction[crypto.Uint256]) []dbft.PublicKey,
@@ -40,6 +40,7 @@ func New(logger *zap.Logger, key dbft.PrivateKey, pub dbft.PublicKey,
 		dbft.WithGetValidators[crypto.Uint256](getValidators),
 		dbft.WithVerifyPrepareRequest[crypto.Uint256](verifyPayload),
 		dbft.WithVerifyPrepareResponse[crypto.Uint256](verifyPayload),
+		dbft.WithVerifyCommit[crypto.Uint256](verifyPayload),
 
 		dbft.WithNewBlockFromContext[crypto.Uint256](newBlockFromContext),
 		dbft.WithNewConsensusPayload[crypto.Uint256](defaultNewConsensusPayload),

internal/simulation/main.go

@@ -179,7 +179,7 @@ func (n *simNode) GetValidators(...dbft.Transaction[crypto.Uint256]) []dbft.Publ
 	return n.validators
 }
 
-func (n *simNode) ProcessBlock(b dbft.Block[crypto.Uint256]) {
+func (n *simNode) ProcessBlock(b dbft.Block[crypto.Uint256]) error {
 	n.d.Logger.Debug("received block", zap.Uint32("height", b.Index()))
 
 	for _, tx := range b.Transactions() {
@@ -188,6 +188,7 @@ func (n *simNode) ProcessBlock(b dbft.Block[crypto.Uint256]) {
 
 	n.height = b.Index()
 	n.lastHash = b.Hash()
+	return nil
 }
 
 // VerifyPayload verifies that payload was received from a good validator.