diff --git a/consensus/state.go b/consensus/state.go index e2e21a3d7..5a4c3c7a3 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -768,7 +768,31 @@ func (cs *State) receiveRoutine(maxSteps int) { // handles proposals, block parts, votes // may generate internal events (votes, complete proposals, 2/3 majorities) - cs.handleMsg(mi) + err := cs.handleMsg(mi) + if err != nil { + cs.Logger.Error( + "failed to process message", + "height", cs.Height, + "round", cs.Round, + "peer", peerID, + "msg_type", fmt.Sprintf("%T", msg), + "err", err, + ) + // if error is due to misbehaviour, disconect from peer + // TODO: can we create some higher order type to collect these errors in? + switch err { + case ErrInvalidProposalSignature, + types.ErrVoteInvalidValidatorIndex, + types.ErrVoteInvalidValidatorAddress, + types.ErrVoteInvalidSignature: + //TODO: disconnect from peer. how best to do this ? options: + // - give ConsensusState a reference to ConR or Switch or StopPeer func + // - introduce new StopPeer event for reactor to listen for + // - use the statsMsgQueue (or a new routine/channel) + default: + //nothing to do + } + } case mi = <-cs.internalMsgQueue: err := cs.wal.WriteSync(mi) // NOTE: fsync @@ -788,7 +812,18 @@ func (cs *State) receiveRoutine(maxSteps int) { } // handles proposals, block parts, votes - cs.handleMsg(mi) + err := cs.handleMsg(mi) + if err != nil { + // TODO: why are we erroring on our own messages? + // should we panic ? + cs.Logger.Error( + "failed to process internal message", + "height", cs.Height, + "round", cs.Round, + "msg_type", fmt.Sprintf("%T", msg), + "err", err, + ) + } case ti := <-cs.timeoutTicker.Chan(): // tockChan: if err := cs.wal.Write(ti); err != nil { @@ -807,13 +842,9 @@ func (cs *State) receiveRoutine(maxSteps int) { } // state transitions on complete-proposal, 2/3-any, 2/3-one -func (cs *State) handleMsg(mi msgInfo) { +func (cs *State) handleMsg(mi msgInfo) error { cs.mtx.Lock() defer cs.mtx.Unlock() - var ( - added bool - err error - ) msg, peerID := mi.Msg, mi.PeerID @@ -821,11 +852,11 @@ func (cs *State) handleMsg(mi msgInfo) { case *ProposalMessage: // will not cause transition. // once proposal is set, we can receive block parts - err = cs.setProposal(msg.Proposal) + return cs.setProposal(msg.Proposal) case *BlockPartMessage: // if the proposal is complete, we'll enterPrevote or tryFinalizeCommit - added, err = cs.addProposalBlockPart(msg, peerID) + added, err := cs.addProposalBlockPart(msg, peerID) // We unlock here to yield to any routines that need to read the the RoundState. // Previously, this code held the lock from the point at which the final block @@ -857,11 +888,12 @@ func (cs *State) handleMsg(mi msgInfo) { ) err = nil } + return err case *VoteMessage: // attempt to add the vote and dupeout the validator if its a duplicate signature // if the vote gives us a 2/3-any or 2/3-one, we transition - added, err = cs.tryAddVote(msg.Vote, peerID) + added, err := cs.tryAddVote(msg.Vote, peerID) if added { cs.statsMsgQueue <- mi } @@ -880,21 +912,11 @@ func (cs *State) handleMsg(mi msgInfo) { // TODO: If rs.Height == vote.Height && rs.Round < vote.Round, // the peer is sending us CatchupCommit precommits. // We could make note of this and help filter in broadcastHasVoteMessage(). + return err default: cs.Logger.Error("unknown msg type", "type", fmt.Sprintf("%T", msg)) - return - } - - if err != nil { - cs.Logger.Error( - "failed to process message", - "height", cs.Height, - "round", cs.Round, - "peer", peerID, - "msg_type", fmt.Sprintf("%T", msg), - "err", err, - ) + return nil } } @@ -1892,9 +1914,9 @@ func (cs *State) defaultSetProposal(proposal *types.Proposal) error { return nil } -// NOTE: block is not necessarily valid. -// Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit, -// once we have the full block. +// Once we have the full block, this will asynchronously trigger either enterPrevote (before we timeout of propose) or tryFinalizeCommit, +// NOTE: block is not necessarily valid. Invalid block means a bad proposer, which +// we can't punish at the peer layer (could be punished by consensus) func (cs *State) addProposalBlockPart(msg *BlockPartMessage, peerID p2p.ID) (added bool, err error) { height, round, part := msg.Height, msg.Round, msg.Part @@ -1922,6 +1944,7 @@ func (cs *State) addProposalBlockPart(msg *BlockPartMessage, peerID p2p.ID) (add added, err = cs.ProposalBlockParts.AddPart(part) if err != nil { + // TODO: can we punish for these or is there a reason they might accidentaly happen? if errors.Is(err, types.ErrPartSetInvalidProof) || errors.Is(err, types.ErrPartSetUnexpectedIndex) { cs.metrics.BlockGossipPartsReceived.With("matches_current", "false").Add(1) }