You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
363 lines
11 KiB
363 lines
11 KiB
package evidence
|
|
|
|
import (
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
clist "github.com/tendermint/tendermint/internal/libs/clist"
|
|
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
|
|
"github.com/tendermint/tendermint/internal/p2p"
|
|
"github.com/tendermint/tendermint/libs/log"
|
|
"github.com/tendermint/tendermint/libs/service"
|
|
tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
var (
|
|
_ service.Service = (*Reactor)(nil)
|
|
|
|
// ChannelShims contains a map of ChannelDescriptorShim objects, where each
|
|
// object wraps a reference to a legacy p2p ChannelDescriptor and the corresponding
|
|
// p2p proto.Message the new p2p Channel is responsible for handling.
|
|
//
|
|
//
|
|
// TODO: Remove once p2p refactor is complete.
|
|
// ref: https://github.com/tendermint/tendermint/issues/5670
|
|
ChannelShims = map[p2p.ChannelID]*p2p.ChannelDescriptorShim{
|
|
EvidenceChannel: {
|
|
MsgType: new(tmproto.EvidenceList),
|
|
Descriptor: &p2p.ChannelDescriptor{
|
|
ID: byte(EvidenceChannel),
|
|
Priority: 6,
|
|
RecvMessageCapacity: maxMsgSize,
|
|
|
|
MaxSendBytes: 400,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
const (
|
|
EvidenceChannel = p2p.ChannelID(0x38)
|
|
|
|
maxMsgSize = 1048576 // 1MB TODO make it configurable
|
|
|
|
// broadcast all uncommitted evidence this often. This sets when the reactor
|
|
// goes back to the start of the list and begins sending the evidence again.
|
|
// Most evidence should be committed in the very next block that is why we wait
|
|
// just over the block production rate before sending evidence again.
|
|
broadcastEvidenceIntervalS = 10
|
|
)
|
|
|
|
// Reactor handles evpool evidence broadcasting amongst peers.
|
|
type Reactor struct {
|
|
service.BaseService
|
|
|
|
evpool *Pool
|
|
evidenceCh *p2p.Channel
|
|
peerUpdates *p2p.PeerUpdates
|
|
closeCh chan struct{}
|
|
|
|
peerWG sync.WaitGroup
|
|
|
|
mtx tmsync.Mutex
|
|
peerRoutines map[types.NodeID]*tmsync.Closer
|
|
}
|
|
|
|
// NewReactor returns a reference to a new evidence reactor, which implements the
|
|
// service.Service interface. It accepts a p2p Channel dedicated for handling
|
|
// envelopes with EvidenceList messages.
|
|
func NewReactor(
|
|
logger log.Logger,
|
|
evidenceCh *p2p.Channel,
|
|
peerUpdates *p2p.PeerUpdates,
|
|
evpool *Pool,
|
|
) *Reactor {
|
|
r := &Reactor{
|
|
evpool: evpool,
|
|
evidenceCh: evidenceCh,
|
|
peerUpdates: peerUpdates,
|
|
closeCh: make(chan struct{}),
|
|
peerRoutines: make(map[types.NodeID]*tmsync.Closer),
|
|
}
|
|
|
|
r.BaseService = *service.NewBaseService(logger, "Evidence", r)
|
|
return r
|
|
}
|
|
|
|
// OnStart starts separate go routines for each p2p Channel and listens for
|
|
// envelopes on each. In addition, it also listens for peer updates and handles
|
|
// messages on that p2p channel accordingly. The caller must be sure to execute
|
|
// OnStop to ensure the outbound p2p Channels are closed. No error is returned.
|
|
func (r *Reactor) OnStart() error {
|
|
go r.processEvidenceCh()
|
|
go r.processPeerUpdates()
|
|
|
|
return nil
|
|
}
|
|
|
|
// OnStop stops the reactor by signaling to all spawned goroutines to exit and
|
|
// blocking until they all exit.
|
|
func (r *Reactor) OnStop() {
|
|
r.mtx.Lock()
|
|
for _, c := range r.peerRoutines {
|
|
c.Close()
|
|
}
|
|
r.mtx.Unlock()
|
|
|
|
// Wait for all spawned peer evidence broadcasting goroutines to gracefully
|
|
// exit.
|
|
r.peerWG.Wait()
|
|
|
|
// Close closeCh to signal to all spawned goroutines to gracefully exit. All
|
|
// p2p Channels should execute Close().
|
|
close(r.closeCh)
|
|
|
|
// Wait for all p2p Channels to be closed before returning. This ensures we
|
|
// can easily reason about synchronization of all p2p Channels and ensure no
|
|
// panics will occur.
|
|
<-r.evidenceCh.Done()
|
|
<-r.peerUpdates.Done()
|
|
}
|
|
|
|
// handleEvidenceMessage handles envelopes sent from peers on the EvidenceChannel.
|
|
// It returns an error only if the Envelope.Message is unknown for this channel
|
|
// or if the given evidence is invalid. This should never be called outside of
|
|
// handleMessage.
|
|
func (r *Reactor) handleEvidenceMessage(envelope p2p.Envelope) error {
|
|
logger := r.Logger.With("peer", envelope.From)
|
|
|
|
switch msg := envelope.Message.(type) {
|
|
case *tmproto.EvidenceList:
|
|
// TODO: Refactor the Evidence type to not contain a list since we only ever
|
|
// send and receive one piece of evidence at a time. Or potentially consider
|
|
// batching evidence.
|
|
//
|
|
// see: https://github.com/tendermint/tendermint/issues/4729
|
|
for i := 0; i < len(msg.Evidence); i++ {
|
|
ev, err := types.EvidenceFromProto(&msg.Evidence[i])
|
|
if err != nil {
|
|
logger.Error("failed to convert evidence", "err", err)
|
|
continue
|
|
}
|
|
|
|
if err := r.evpool.AddEvidence(ev); err != nil {
|
|
// If we're given invalid evidence by the peer, notify the router that
|
|
// we should remove this peer by returning an error.
|
|
if _, ok := err.(*types.ErrInvalidEvidence); ok {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
default:
|
|
return fmt.Errorf("received unknown message: %T", msg)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
|
|
// It will handle errors and any possible panics gracefully. A caller can handle
|
|
// any error returned by sending a PeerError on the respective channel.
|
|
func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
|
|
defer func() {
|
|
if e := recover(); e != nil {
|
|
err = fmt.Errorf("panic in processing message: %v", e)
|
|
}
|
|
}()
|
|
|
|
r.Logger.Debug("received message", "message", envelope.Message, "peer", envelope.From)
|
|
|
|
switch chID {
|
|
case EvidenceChannel:
|
|
err = r.handleEvidenceMessage(envelope)
|
|
|
|
default:
|
|
err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// processEvidenceCh implements a blocking event loop where we listen for p2p
|
|
// Envelope messages from the evidenceCh.
|
|
func (r *Reactor) processEvidenceCh() {
|
|
defer r.evidenceCh.Close()
|
|
|
|
for {
|
|
select {
|
|
case envelope := <-r.evidenceCh.In:
|
|
if err := r.handleMessage(r.evidenceCh.ID, envelope); err != nil {
|
|
r.Logger.Error("failed to process message", "ch_id", r.evidenceCh.ID, "envelope", envelope, "err", err)
|
|
r.evidenceCh.Error <- p2p.PeerError{
|
|
NodeID: envelope.From,
|
|
Err: err,
|
|
}
|
|
}
|
|
|
|
case <-r.closeCh:
|
|
r.Logger.Debug("stopped listening on evidence channel; closing...")
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// processPeerUpdate processes a PeerUpdate. For new or live peers it will check
|
|
// if an evidence broadcasting goroutine needs to be started. For down or
|
|
// removed peers, it will check if an evidence broadcasting goroutine
|
|
// exists and signal that it should exit.
|
|
//
|
|
// FIXME: The peer may be behind in which case it would simply ignore the
|
|
// evidence and treat it as invalid. This would cause the peer to disconnect.
|
|
// The peer may also receive the same piece of evidence multiple times if it
|
|
// connects/disconnects frequently from the broadcasting peer(s).
|
|
//
|
|
// REF: https://github.com/tendermint/tendermint/issues/4727
|
|
func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
|
|
r.Logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
|
|
|
|
r.mtx.Lock()
|
|
defer r.mtx.Unlock()
|
|
|
|
switch peerUpdate.Status {
|
|
case p2p.PeerStatusUp:
|
|
// Do not allow starting new evidence broadcast loops after reactor shutdown
|
|
// has been initiated. This can happen after we've manually closed all
|
|
// peer broadcast loops and closed r.closeCh, but the router still sends
|
|
// in-flight peer updates.
|
|
if !r.IsRunning() {
|
|
return
|
|
}
|
|
|
|
// Check if we've already started a goroutine for this peer, if not we create
|
|
// a new done channel so we can explicitly close the goroutine if the peer
|
|
// is later removed, we increment the waitgroup so the reactor can stop
|
|
// safely, and finally start the goroutine to broadcast evidence to that peer.
|
|
_, ok := r.peerRoutines[peerUpdate.NodeID]
|
|
if !ok {
|
|
closer := tmsync.NewCloser()
|
|
|
|
r.peerRoutines[peerUpdate.NodeID] = closer
|
|
r.peerWG.Add(1)
|
|
go r.broadcastEvidenceLoop(peerUpdate.NodeID, closer)
|
|
}
|
|
|
|
case p2p.PeerStatusDown:
|
|
// Check if we've started an evidence broadcasting goroutine for this peer.
|
|
// If we have, we signal to terminate the goroutine via the channel's closure.
|
|
// This will internally decrement the peer waitgroup and remove the peer
|
|
// from the map of peer evidence broadcasting goroutines.
|
|
closer, ok := r.peerRoutines[peerUpdate.NodeID]
|
|
if ok {
|
|
closer.Close()
|
|
}
|
|
}
|
|
}
|
|
|
|
// processPeerUpdates initiates a blocking process where we listen for and handle
|
|
// PeerUpdate messages. When the reactor is stopped, we will catch the signal and
|
|
// close the p2p PeerUpdatesCh gracefully.
|
|
func (r *Reactor) processPeerUpdates() {
|
|
defer r.peerUpdates.Close()
|
|
|
|
for {
|
|
select {
|
|
case peerUpdate := <-r.peerUpdates.Updates():
|
|
r.processPeerUpdate(peerUpdate)
|
|
|
|
case <-r.closeCh:
|
|
r.Logger.Debug("stopped listening on peer updates channel; closing...")
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// broadcastEvidenceLoop starts a blocking process that continuously reads pieces
|
|
// of evidence off of a linked-list and sends the evidence in a p2p Envelope to
|
|
// the given peer by ID. This should be invoked in a goroutine per unique peer
|
|
// ID via an appropriate PeerUpdate. The goroutine can be signaled to gracefully
|
|
// exit by either explicitly closing the provided doneCh or by the reactor
|
|
// signaling to stop.
|
|
//
|
|
// TODO: This should be refactored so that we do not blindly gossip evidence
|
|
// that the peer has already received or may not be ready for.
|
|
//
|
|
// REF: https://github.com/tendermint/tendermint/issues/4727
|
|
func (r *Reactor) broadcastEvidenceLoop(peerID types.NodeID, closer *tmsync.Closer) {
|
|
var next *clist.CElement
|
|
|
|
defer func() {
|
|
r.mtx.Lock()
|
|
delete(r.peerRoutines, peerID)
|
|
r.mtx.Unlock()
|
|
|
|
r.peerWG.Done()
|
|
|
|
if e := recover(); e != nil {
|
|
r.Logger.Error("recovering from broadcasting evidence loop", "err", e)
|
|
}
|
|
}()
|
|
|
|
for {
|
|
// This happens because the CElement we were looking at got garbage
|
|
// collected (removed). That is, .NextWaitChan() returned nil. So we can go
|
|
// ahead and start from the beginning.
|
|
if next == nil {
|
|
select {
|
|
case <-r.evpool.EvidenceWaitChan(): // wait until next evidence is available
|
|
if next = r.evpool.EvidenceFront(); next == nil {
|
|
continue
|
|
}
|
|
|
|
case <-closer.Done():
|
|
// The peer is marked for removal via a PeerUpdate as the doneCh was
|
|
// explicitly closed to signal we should exit.
|
|
return
|
|
|
|
case <-r.closeCh:
|
|
// The reactor has signaled that we are stopped and thus we should
|
|
// implicitly exit this peer's goroutine.
|
|
return
|
|
}
|
|
}
|
|
|
|
ev := next.Value.(types.Evidence)
|
|
evProto, err := types.EvidenceToProto(ev)
|
|
if err != nil {
|
|
panic(fmt.Errorf("failed to convert evidence: %w", err))
|
|
}
|
|
|
|
// Send the evidence to the corresponding peer. Note, the peer may be behind
|
|
// and thus would not be able to process the evidence correctly. Also, the
|
|
// peer may receive this piece of evidence multiple times if it added and
|
|
// removed frequently from the broadcasting peer.
|
|
r.evidenceCh.Out <- p2p.Envelope{
|
|
To: peerID,
|
|
Message: &tmproto.EvidenceList{
|
|
Evidence: []tmproto.Evidence{*evProto},
|
|
},
|
|
}
|
|
r.Logger.Debug("gossiped evidence to peer", "evidence", ev, "peer", peerID)
|
|
|
|
select {
|
|
case <-time.After(time.Second * broadcastEvidenceIntervalS):
|
|
// start from the beginning after broadcastEvidenceIntervalS seconds
|
|
next = nil
|
|
|
|
case <-next.NextWaitChan():
|
|
next = next.Next()
|
|
|
|
case <-closer.Done():
|
|
// The peer is marked for removal via a PeerUpdate as the doneCh was
|
|
// explicitly closed to signal we should exit.
|
|
return
|
|
|
|
case <-r.closeCh:
|
|
// The reactor has signaled that we are stopped and thus we should
|
|
// implicitly exit this peer's goroutine.
|
|
return
|
|
}
|
|
}
|
|
}
|