Move blockpart redundancy checks to reactor

.changelog/unreleased/improvements/3161-move-blockpart-redundancy-checks-into-reactor.md

@@ -0,0 +1,2 @@
+- [`consensus`] Move blockpart redundancy/lateness checks into the reactor, preventing late / duplicate parts from blocking consensus.
+  ([\#3161](https://github.com/cometbft/cometbft/issues/3161)

consensus/reactor.go

@@ -332,6 +332,17 @@ func (conR *Reactor) Receive(e p2p.Envelope) {
 		case *BlockPartMessage:
 			ps.SetHasProposalBlockPart(msg.Height, msg.Round, int(msg.Part.Index))
 			conR.Metrics.BlockParts.With("peer_id", string(e.Src.ID())).Add(1)
+
+			conR.rsMtx.RLock()
+			height, blockParts := conR.rs.Height, conR.rs.ProposalBlockParts
+			conR.rsMtx.RUnlock()
+
+			allowFutureBlockPart := true
+			ok := allowProcessingProposalBlockPart(msg, conR.Logger, conR.Metrics, height, blockParts, allowFutureBlockPart, e.Src.ID())
+			if !ok {
+				return
+			}
+
 			conR.conS.peerMsgQueue <- msgInfo{msg, e.Src.ID()}
 		default:
 			conR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))

consensus/state.go

@@ -1979,31 +1979,62 @@ func (cs *State) readSerializedBlockFromBlockParts() ([]byte, error) {
 	return serializedBlockBuffer, nil
 }
 
-// NOTE: block is not necessarily valid.
-// Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit,
-// once we have the full block.
-func (cs *State) addProposalBlockPart(msg *BlockPartMessage, peerID p2p.ID) (added bool, err error) {
+// checks if we should allow processing the proposal block part.
+// Shared code between reactor and state machine.
+// This must not modify csBlockParts, only take read-only accesses to it.
+// Returns true if the block part is not old or duplicated.
+func allowProcessingProposalBlockPart(msg *BlockPartMessage, logger log.Logger, metrics *Metrics, csHeight int64, csBlockParts *types.PartSet, allowFutureHeights bool, peerID p2p.ID) bool {
 	height, round, part := msg.Height, msg.Round, msg.Part
 
-	// Blocks might be reused, so round mismatch is OK
-	if cs.Height != height {
-		cs.Logger.Debug("received block part from wrong height", "height", height, "round", round)
-		cs.metrics.BlockGossipPartsReceived.With("matches_current", "false").Add(1)
-		return false, nil
+	// Blocks might be reused, so round mismatch is OK. Meant for reactor, where we may get
+	// future block parts while the proposal for the next block is still in message queue.
+	if allowFutureHeights && height > csHeight {
+		return true
+	}
+	if csHeight != height {
+		logger.Debug("received block part from wrong height", "height", height, "round", round)
+		metrics.BlockGossipPartsReceived.With("matches_current", "false").Add(1)
+		return false
 	}
 
 	// We're not expecting a block part.
-	if cs.ProposalBlockParts == nil {
-		cs.metrics.BlockGossipPartsReceived.With("matches_current", "false").Add(1)
+	if csBlockParts == nil {
+		if allowFutureHeights {
+			return true
+		}
+		metrics.BlockGossipPartsReceived.With("matches_current", "false").Add(1)
 		// NOTE: this can happen when we've gone to a higher round and
 		// then receive parts from the previous round - not necessarily a bad peer.
-		cs.Logger.Debug(
+		logger.Debug(
 			"received a block part when we are not expecting any",
 			"height", height,
 			"round", round,
 			"index", part.Index,
 			"peer", peerID,
 		)
+		return false
+	}
+
+	if part.Index >= csBlockParts.Total() {
+		return false
+	}
+
+	if csBlockParts.IsCompleteMtx() || csBlockParts.GetPart(int(part.Index)) != nil {
+		return false
+	}
+
+	return true
+}
+
+// NOTE: block is not necessarily valid.
+// Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit,
+// once we have the full block.
+func (cs *State) addProposalBlockPart(msg *BlockPartMessage, peerID p2p.ID) (added bool, err error) {
+	part := msg.Part
+	// TODO: better handle block parts for future heights, by saving them and processing them later.
+	allowFutureBlockPart := false
+	ok := allowProcessingProposalBlockPart(msg, cs.Logger, cs.metrics, cs.Height, cs.ProposalBlockParts, allowFutureBlockPart, peerID)
+	if !ok {
 		return false, nil
 	}
 

types/part_set.go

@@ -322,6 +322,12 @@ func (ps *PartSet) IsComplete() bool {
 	return ps.count == ps.total
 }
 
+func (ps *PartSet) IsCompleteMtx() bool {
+	ps.mtx.Lock()
+	defer ps.mtx.Unlock()
+	return ps.count == ps.total
+}
+
 func (ps *PartSet) GetReader() io.Reader {
 	if !ps.IsComplete() {
 		panic("Cannot GetReader() on incomplete PartSet")