add receives to all with NewReceive

blockchain/v1/reactor.go

@@ -0,0 +1,550 @@
+package v1
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/tendermint/tendermint/behaviour"
+	bc "github.com/tendermint/tendermint/blockchain"
+	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/p2p"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	sm "github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/store"
+	"github.com/tendermint/tendermint/types"
+)
+
+const (
+	// BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
+	BlockchainChannel = byte(0x40)
+	trySyncIntervalMS = 10
+	trySendIntervalMS = 10
+
+	// ask for best height every 10s
+	statusUpdateIntervalSeconds = 10
+)
+
+var (
+	// Maximum number of requests that can be pending per peer, i.e. for which requests have been sent but blocks
+	// have not been received.
+	maxRequestsPerPeer = 20
+	// Maximum number of block requests for the reactor, pending or for which blocks have been received.
+	maxNumRequests = 64
+)
+
+type consensusReactor interface {
+	// for when we switch from blockchain reactor and fast sync to
+	// the consensus machine
+	SwitchToConsensus(state sm.State, skipWAL bool)
+}
+
+// BlockchainReactor handles long-term catchup syncing.
+type BlockchainReactor struct {
+	p2p.BaseReactor
+
+	initialState sm.State // immutable
+	state        sm.State
+
+	blockExec *sm.BlockExecutor
+	store     *store.BlockStore
+
+	fastSync    bool
+	stateSynced bool
+
+	fsm          *BcReactorFSM
+	blocksSynced uint64
+
+	// Receive goroutine forwards messages to this channel to be processed in the context of the poolRoutine.
+	messagesForFSMCh chan bcReactorMessage
+
+	// Switch goroutine may send RemovePeer to the blockchain reactor. This is an error message that is relayed
+	// to this channel to be processed in the context of the poolRoutine.
+	errorsForFSMCh chan bcReactorMessage
+
+	// This channel is used by the FSM and indirectly the block pool to report errors to the blockchain reactor and
+	// the switch.
+	eventsFromFSMCh chan bcFsmMessage
+
+	swReporter *behaviour.SwitchReporter
+}
+
+// NewBlockchainReactor returns new reactor instance.
+func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore,
+	fastSync bool) *BlockchainReactor {
+
+	if state.LastBlockHeight != store.Height() {
+		panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight,
+			store.Height()))
+	}
+
+	const capacity = 1000
+	eventsFromFSMCh := make(chan bcFsmMessage, capacity)
+	messagesForFSMCh := make(chan bcReactorMessage, capacity)
+	errorsForFSMCh := make(chan bcReactorMessage, capacity)
+
+	startHeight := store.Height() + 1
+	if startHeight == 1 {
+		startHeight = state.InitialHeight
+	}
+	bcR := &BlockchainReactor{
+		initialState:     state,
+		state:            state,
+		blockExec:        blockExec,
+		fastSync:         fastSync,
+		store:            store,
+		messagesForFSMCh: messagesForFSMCh,
+		eventsFromFSMCh:  eventsFromFSMCh,
+		errorsForFSMCh:   errorsForFSMCh,
+	}
+	fsm := NewFSM(startHeight, bcR)
+	bcR.fsm = fsm
+	bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
+	// bcR.swReporter = behaviour.NewSwitchReporter(bcR.BaseReactor.Switch)
+
+	return bcR
+}
+
+// bcReactorMessage is used by the reactor to send messages to the FSM.
+type bcReactorMessage struct {
+	event bReactorEvent
+	data  bReactorEventData
+}
+
+type bFsmEvent uint
+
+const (
+	// message type events
+	peerErrorEv = iota + 1
+	syncFinishedEv
+)
+
+type bFsmEventData struct {
+	peerID p2p.ID
+	err    error
+}
+
+// bcFsmMessage is used by the FSM to send messages to the reactor
+type bcFsmMessage struct {
+	event bFsmEvent
+	data  bFsmEventData
+}
+
+// SetLogger implements service.Service by setting the logger on reactor and pool.
+func (bcR *BlockchainReactor) SetLogger(l log.Logger) {
+	bcR.BaseService.Logger = l
+	bcR.fsm.SetLogger(l)
+}
+
+// OnStart implements service.Service.
+func (bcR *BlockchainReactor) OnStart() error {
+	bcR.swReporter = behaviour.NewSwitchReporter(bcR.BaseReactor.Switch)
+	if bcR.fastSync {
+		go bcR.poolRoutine()
+	}
+	return nil
+}
+
+// OnStop implements service.Service.
+func (bcR *BlockchainReactor) OnStop() {
+	_ = bcR.Stop()
+}
+
+// SwitchToFastSync is called by the state sync reactor when switching to fast sync.
+func (bcR *BlockchainReactor) SwitchToFastSync(state sm.State) error {
+	bcR.fastSync = true
+	bcR.initialState = state
+	bcR.state = state
+	bcR.stateSynced = true
+
+	bcR.fsm = NewFSM(state.LastBlockHeight+1, bcR)
+	bcR.fsm.SetLogger(bcR.Logger)
+	go bcR.poolRoutine()
+	return nil
+}
+
+// GetChannels implements Reactor
+func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
+	return []*p2p.ChannelDescriptor{
+		{
+			ID:                  BlockchainChannel,
+			Priority:            10,
+			SendQueueCapacity:   2000,
+			RecvBufferCapacity:  50 * 4096,
+			RecvMessageCapacity: bc.MaxMsgSize,
+		},
+	}
+}
+
+// AddPeer implements Reactor by sending our state to peer.
+func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) {
+	peer.NewSend(p2p.Envelope{
+		ChannelID: BlockchainChannel,
+		Message: &bcproto.StatusResponse{
+			Base:   bcR.store.Base(),
+			Height: bcR.store.Height(),
+		},
+	})
+	// it's OK if send fails. will try later in poolRoutine
+
+	// peer is added to the pool once we receive the first
+	// bcStatusResponseMessage from the peer and call pool.updatePeer()
+}
+
+// sendBlockToPeer loads a block and sends it to the requesting peer.
+// If the block doesn't exist a bcNoBlockResponseMessage is sent.
+// If all nodes are honest, no node should be requesting for a block that doesn't exist.
+func (bcR *BlockchainReactor) sendBlockToPeer(msg *bcproto.BlockRequest,
+	src p2p.Peer) (queued bool) {
+
+	block := bcR.store.LoadBlock(msg.Height)
+	if block != nil {
+		pbbi, err := block.ToProto()
+		if err != nil {
+			bcR.Logger.Error("Could not send block message to peer", "err", err)
+			return false
+		}
+		return src.NewTrySend(p2p.Envelope{
+			ChannelID: BlockchainChannel,
+			Message:   &bcproto.BlockResponse{Block: pbbi},
+		})
+	}
+
+	bcR.Logger.Info("peer asking for a block we don't have", "src", src, "height", msg.Height)
+
+	return src.NewTrySend(p2p.Envelope{
+		ChannelID: BlockchainChannel,
+		Message:   &bcproto.NoBlockResponse{Height: msg.Height},
+	})
+}
+
+func (bcR *BlockchainReactor) sendStatusResponseToPeer(msg *bcproto.StatusRequest, src p2p.Peer) (queued bool) {
+	return src.NewTrySend(p2p.Envelope{
+		ChannelID: BlockchainChannel,
+		Message: &bcproto.StatusResponse{
+			Base:   bcR.store.Base(),
+			Height: bcR.store.Height(),
+		},
+	})
+}
+
+// RemovePeer implements Reactor by removing peer from the pool.
+func (bcR *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
+	msgData := bcReactorMessage{
+		event: peerRemoveEv,
+		data: bReactorEventData{
+			peerID: peer.ID(),
+			err:    errSwitchRemovesPeer,
+		},
+	}
+	bcR.errorsForFSMCh <- msgData
+}
+
+// Receive implements Reactor by handling 4 types of messages (look below).
+func (bcR *BlockchainReactor) NewReceive(e p2p.Envelope) {
+	if err := bc.ValidateMsg(e.Message); err != nil {
+		bcR.Logger.Error("peer sent us invalid msg", "peer", e.Src, "msg", e.Message, "err", err)
+		_ = bcR.swReporter.Report(behaviour.BadMessage(e.Src.ID(), err.Error()))
+		return
+	}
+
+	bcR.Logger.Debug("Receive", "src", e.Src, "chID", e.ChannelID, "msg", e.Message)
+
+	switch msg := e.Message.(type) {
+	case *bcproto.BlockRequest:
+		if queued := bcR.sendBlockToPeer(msg, e.Src); !queued {
+			// Unfortunately not queued since the queue is full.
+			bcR.Logger.Error("Could not send block message to peer", "src", e.Src, "height", msg.Height)
+		}
+
+	case *bcproto.StatusRequest:
+		// Send peer our state.
+		if queued := bcR.sendStatusResponseToPeer(msg, e.Src); !queued {
+			// Unfortunately not queued since the queue is full.
+			bcR.Logger.Error("Could not send status message to peer", "src", e.Src)
+		}
+
+	case *bcproto.BlockResponse:
+		bi, err := types.BlockFromProto(msg.Block)
+		if err != nil {
+			bcR.Logger.Error("error transition block from protobuf", "err", err)
+			return
+		}
+		msgForFSM := bcReactorMessage{
+			event: blockResponseEv,
+			data: bReactorEventData{
+				peerID: e.Src.ID(),
+				height: bi.Height,
+				block:  bi,
+			},
+		}
+		bcR.Logger.Info("Received", "src", e.Src, "height", bi.Height)
+		bcR.messagesForFSMCh <- msgForFSM
+	case *bcproto.NoBlockResponse:
+		msgForFSM := bcReactorMessage{
+			event: noBlockResponseEv,
+			data: bReactorEventData{
+				peerID: e.Src.ID(),
+				height: msg.Height,
+			},
+		}
+		bcR.Logger.Debug("Peer does not have requested block", "peer", e.Src, "height", msg.Height)
+		bcR.messagesForFSMCh <- msgForFSM
+
+	case *bcproto.StatusResponse:
+		// Got a peer status. Unverified.
+		msgForFSM := bcReactorMessage{
+			event: statusResponseEv,
+			data: bReactorEventData{
+				peerID: e.Src.ID(),
+				height: msg.Height,
+			},
+		}
+		bcR.messagesForFSMCh <- msgForFSM
+
+	default:
+		bcR.Logger.Error(fmt.Sprintf("unknown message type %T", msg))
+	}
+}
+
+func (bcR *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
+// processBlocksRoutine processes blocks until signlaed to stop over the stopProcessing channel
+func (bcR *BlockchainReactor) processBlocksRoutine(stopProcessing chan struct{}) {
+
+	processReceivedBlockTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
+	doProcessBlockCh := make(chan struct{}, 1)
+
+	lastHundred := time.Now()
+	lastRate := 0.0
+
+ForLoop:
+	for {
+		select {
+		case <-stopProcessing:
+			bcR.Logger.Info("finishing block execution")
+			break ForLoop
+		case <-processReceivedBlockTicker.C: // try to execute blocks
+			select {
+			case doProcessBlockCh <- struct{}{}:
+			default:
+			}
+		case <-doProcessBlockCh:
+			for {
+				err := bcR.processBlock()
+				if err == errMissingBlock {
+					break
+				}
+				// Notify FSM of block processing result.
+				msgForFSM := bcReactorMessage{
+					event: processedBlockEv,
+					data: bReactorEventData{
+						err: err,
+					},
+				}
+				_ = bcR.fsm.Handle(&msgForFSM)
+
+				if err != nil {
+					break
+				}
+
+				bcR.blocksSynced++
+				if bcR.blocksSynced%100 == 0 {
+					lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
+					height, maxPeerHeight := bcR.fsm.Status()
+					bcR.Logger.Info("Fast Sync Rate", "height", height,
+						"max_peer_height", maxPeerHeight, "blocks/s", lastRate)
+					lastHundred = time.Now()
+				}
+			}
+		}
+	}
+}
+
+// poolRoutine receives and handles messages from the Receive() routine and from the FSM.
+func (bcR *BlockchainReactor) poolRoutine() {
+
+	bcR.fsm.Start()
+
+	sendBlockRequestTicker := time.NewTicker(trySendIntervalMS * time.Millisecond)
+	statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
+
+	stopProcessing := make(chan struct{}, 1)
+	go bcR.processBlocksRoutine(stopProcessing)
+
+ForLoop:
+	for {
+		select {
+
+		case <-sendBlockRequestTicker.C:
+			if !bcR.fsm.NeedsBlocks() {
+				continue
+			}
+			_ = bcR.fsm.Handle(&bcReactorMessage{
+				event: makeRequestsEv,
+				data: bReactorEventData{
+					maxNumRequests: maxNumRequests}})
+
+		case <-statusUpdateTicker.C:
+			// Ask for status updates.
+			go bcR.sendStatusRequest()
+
+		case msg := <-bcR.messagesForFSMCh:
+			// Sent from the Receive() routine when status (statusResponseEv) and
+			// block (blockResponseEv) response events are received
+			_ = bcR.fsm.Handle(&msg)
+
+		case msg := <-bcR.errorsForFSMCh:
+			// Sent from the switch.RemovePeer() routine (RemovePeerEv) and
+			// FSM state timer expiry routine (stateTimeoutEv).
+			_ = bcR.fsm.Handle(&msg)
+
+		case msg := <-bcR.eventsFromFSMCh:
+			switch msg.event {
+			case syncFinishedEv:
+				stopProcessing <- struct{}{}
+				// Sent from the FSM when it enters finished state.
+				break ForLoop
+			case peerErrorEv:
+				// Sent from the FSM when it detects peer error
+				bcR.reportPeerErrorToSwitch(msg.data.err, msg.data.peerID)
+				if msg.data.err == errNoPeerResponse {
+					// Sent from the peer timeout handler routine
+					_ = bcR.fsm.Handle(&bcReactorMessage{
+						event: peerRemoveEv,
+						data: bReactorEventData{
+							peerID: msg.data.peerID,
+							err:    msg.data.err,
+						},
+					})
+				}
+				// else {
+				// For slow peers, or errors due to blocks received from wrong peer
+				// the FSM had already removed the peers
+				// }
+			default:
+				bcR.Logger.Error("Event from FSM not supported", "type", msg.event)
+			}
+
+		case <-bcR.Quit():
+			break ForLoop
+		}
+	}
+}
+
+func (bcR *BlockchainReactor) reportPeerErrorToSwitch(err error, peerID p2p.ID) {
+	peer := bcR.Switch.Peers().Get(peerID)
+	if peer != nil {
+		_ = bcR.swReporter.Report(behaviour.BadMessage(peerID, err.Error()))
+	}
+}
+
+func (bcR *BlockchainReactor) processBlock() error {
+
+	first, second, err := bcR.fsm.FirstTwoBlocks()
+	if err != nil {
+		// We need both to sync the first block.
+		return err
+	}
+
+	chainID := bcR.initialState.ChainID
+
+	firstParts := first.MakePartSet(types.BlockPartSizeBytes)
+	firstPartSetHeader := firstParts.Header()
+	firstID := types.BlockID{Hash: first.Hash(), PartSetHeader: firstPartSetHeader}
+	// Finally, verify the first block using the second's commit
+	// NOTE: we can probably make this more efficient, but note that calling
+	// first.Hash() doesn't verify the tx contents, so MakePartSet() is
+	// currently necessary.
+	err = bcR.state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit)
+	if err != nil {
+		bcR.Logger.Error("error during commit verification", "err", err,
+			"first", first.Height, "second", second.Height)
+		return errBlockVerificationFailure
+	}
+
+	bcR.store.SaveBlock(first, firstParts, second.LastCommit)
+
+	bcR.state, _, err = bcR.blockExec.ApplyBlock(bcR.state, firstID, first)
+	if err != nil {
+		panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err))
+	}
+
+	return nil
+}
+
+// Implements bcRNotifier
+// sendStatusRequest broadcasts `BlockStore` height.
+func (bcR *BlockchainReactor) sendStatusRequest() {
+	bcR.Switch.NewBroadcast(p2p.Envelope{
+		ChannelID: BlockchainChannel,
+		Message:   &bcproto.StatusRequest{},
+	})
+}
+
+// Implements bcRNotifier
+// BlockRequest sends `BlockRequest` height.
+func (bcR *BlockchainReactor) sendBlockRequest(peerID p2p.ID, height int64) error {
+	peer := bcR.Switch.Peers().Get(peerID)
+	if peer == nil {
+		return errNilPeerForBlockRequest
+	}
+
+	queued := peer.NewTrySend(p2p.Envelope{
+		ChannelID: BlockchainChannel,
+		Message:   &bcproto.BlockRequest{Height: height},
+	})
+	if !queued {
+		return errSendQueueFull
+	}
+	return nil
+}
+
+// Implements bcRNotifier
+func (bcR *BlockchainReactor) switchToConsensus() {
+	conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
+	if ok {
+		conR.SwitchToConsensus(bcR.state, bcR.blocksSynced > 0 || bcR.stateSynced)
+		bcR.eventsFromFSMCh <- bcFsmMessage{event: syncFinishedEv}
+	}
+	// else {
+	// Should only happen during testing.
+	// }
+}
+
+// Implements bcRNotifier
+// Called by FSM and pool:
+// - pool calls when it detects slow peer or when peer times out
+// - FSM calls when:
+//   - adding a block (addBlock) fails
+//   - reactor processing of a block reports failure and FSM sends back the peers of first and second blocks
+func (bcR *BlockchainReactor) sendPeerError(err error, peerID p2p.ID) {
+	bcR.Logger.Info("sendPeerError:", "peer", peerID, "error", err)
+	msgData := bcFsmMessage{
+		event: peerErrorEv,
+		data: bFsmEventData{
+			peerID: peerID,
+			err:    err,
+		},
+	}
+	bcR.eventsFromFSMCh <- msgData
+}
+
+// Implements bcRNotifier
+func (bcR *BlockchainReactor) resetStateTimer(name string, timer **time.Timer, timeout time.Duration) {
+	if timer == nil {
+		panic("nil timer pointer parameter")
+	}
+	if *timer == nil {
+		*timer = time.AfterFunc(timeout, func() {
+			msg := bcReactorMessage{
+				event: stateTimeoutEv,
+				data: bReactorEventData{
+					stateName: name,
+				},
+			}
+			bcR.errorsForFSMCh <- msg
+		})
+	} else {
+		(*timer).Reset(timeout)
+	}
+}

blockchain/v2/reactor.go

@@ -0,0 +1,554 @@
+package v2
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/tendermint/tendermint/behaviour"
+	bc "github.com/tendermint/tendermint/blockchain"
+	"github.com/tendermint/tendermint/libs/log"
+	tmsync "github.com/tendermint/tendermint/libs/sync"
+	"github.com/tendermint/tendermint/p2p"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	"github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/types"
+)
+
+const (
+	// chBufferSize is the buffer size of all event channels.
+	chBufferSize int = 1000
+)
+
+type blockStore interface {
+	LoadBlock(height int64) *types.Block
+	SaveBlock(*types.Block, *types.PartSet, *types.Commit)
+	Base() int64
+	Height() int64
+}
+
+// BlockchainReactor handles fast sync protocol.
+type BlockchainReactor struct {
+	p2p.BaseReactor
+
+	fastSync    bool // if true, enable fast sync on start
+	stateSynced bool // set to true when SwitchToFastSync is called by state sync
+	scheduler   *Routine
+	processor   *Routine
+	logger      log.Logger
+
+	mtx           tmsync.RWMutex
+	maxPeerHeight int64
+	syncHeight    int64
+	events        chan Event // non-nil during a fast sync
+
+	reporter behaviour.Reporter
+	io       iIO
+	store    blockStore
+}
+
+//nolint:unused,deadcode
+type blockVerifier interface {
+	VerifyCommit(chainID string, blockID types.BlockID, height int64, commit *types.Commit) error
+}
+
+type blockApplier interface {
+	ApplyBlock(state state.State, blockID types.BlockID, block *types.Block) (state.State, int64, error)
+}
+
+// XXX: unify naming in this package around tmState
+func newReactor(state state.State, store blockStore, reporter behaviour.Reporter,
+	blockApplier blockApplier, fastSync bool) *BlockchainReactor {
+	initHeight := state.LastBlockHeight + 1
+	if initHeight == 1 {
+		initHeight = state.InitialHeight
+	}
+	scheduler := newScheduler(initHeight, time.Now())
+	pContext := newProcessorContext(store, blockApplier, state)
+	// TODO: Fix naming to just newProcesssor
+	// newPcState requires a processorContext
+	processor := newPcState(pContext)
+
+	return &BlockchainReactor{
+		scheduler: newRoutine("scheduler", scheduler.handle, chBufferSize),
+		processor: newRoutine("processor", processor.handle, chBufferSize),
+		store:     store,
+		reporter:  reporter,
+		logger:    log.NewNopLogger(),
+		fastSync:  fastSync,
+	}
+}
+
+// NewBlockchainReactor creates a new reactor instance.
+func NewBlockchainReactor(
+	state state.State,
+	blockApplier blockApplier,
+	store blockStore,
+	fastSync bool) *BlockchainReactor {
+	reporter := behaviour.NewMockReporter()
+	return newReactor(state, store, reporter, blockApplier, fastSync)
+}
+
+// SetSwitch implements Reactor interface.
+func (r *BlockchainReactor) SetSwitch(sw *p2p.Switch) {
+	r.Switch = sw
+	if sw != nil {
+		r.io = newSwitchIo(sw)
+	} else {
+		r.io = nil
+	}
+}
+
+func (r *BlockchainReactor) setMaxPeerHeight(height int64) {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+	if height > r.maxPeerHeight {
+		r.maxPeerHeight = height
+	}
+}
+
+func (r *BlockchainReactor) setSyncHeight(height int64) {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+	r.syncHeight = height
+}
+
+// SyncHeight returns the height to which the BlockchainReactor has synced.
+func (r *BlockchainReactor) SyncHeight() int64 {
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+	return r.syncHeight
+}
+
+// SetLogger sets the logger of the reactor.
+func (r *BlockchainReactor) SetLogger(logger log.Logger) {
+	r.logger = logger
+	r.scheduler.setLogger(logger)
+	r.processor.setLogger(logger)
+}
+
+// Start implements cmn.Service interface
+func (r *BlockchainReactor) Start() error {
+	r.reporter = behaviour.NewSwitchReporter(r.BaseReactor.Switch)
+	if r.fastSync {
+		err := r.startSync(nil)
+		if err != nil {
+			return fmt.Errorf("failed to start fast sync: %w", err)
+		}
+	}
+	return nil
+}
+
+// startSync begins a fast sync, signalled by r.events being non-nil. If state is non-nil,
+// the scheduler and processor is updated with this state on startup.
+func (r *BlockchainReactor) startSync(state *state.State) error {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+	if r.events != nil {
+		return errors.New("fast sync already in progress")
+	}
+	r.events = make(chan Event, chBufferSize)
+	go r.scheduler.start()
+	go r.processor.start()
+	if state != nil {
+		<-r.scheduler.ready()
+		<-r.processor.ready()
+		r.scheduler.send(bcResetState{state: *state})
+		r.processor.send(bcResetState{state: *state})
+	}
+	go r.demux(r.events)
+	return nil
+}
+
+// endSync ends a fast sync
+func (r *BlockchainReactor) endSync() {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+	if r.events != nil {
+		close(r.events)
+	}
+	r.events = nil
+	r.scheduler.stop()
+	r.processor.stop()
+}
+
+// SwitchToFastSync is called by the state sync reactor when switching to fast sync.
+func (r *BlockchainReactor) SwitchToFastSync(state state.State) error {
+	r.stateSynced = true
+	state = state.Copy()
+	return r.startSync(&state)
+}
+
+// reactor generated ticker events:
+// ticker for cleaning peers
+type rTryPrunePeer struct {
+	priorityHigh
+	time time.Time
+}
+
+func (e rTryPrunePeer) String() string {
+	return fmt.Sprintf("rTryPrunePeer{%v}", e.time)
+}
+
+// ticker event for scheduling block requests
+type rTrySchedule struct {
+	priorityHigh
+	time time.Time
+}
+
+func (e rTrySchedule) String() string {
+	return fmt.Sprintf("rTrySchedule{%v}", e.time)
+}
+
+// ticker for block processing
+type rProcessBlock struct {
+	priorityNormal
+}
+
+func (e rProcessBlock) String() string {
+	return "rProcessBlock"
+}
+
+// reactor generated events based on blockchain related messages from peers:
+// blockResponse message received from a peer
+type bcBlockResponse struct {
+	priorityNormal
+	time   time.Time
+	peerID p2p.ID
+	block  *types.Block
+}
+
+func (resp bcBlockResponse) String() string {
+	return fmt.Sprintf("bcBlockResponse{%d#%X from %v at %v}",
+		resp.block.Height, resp.block.Hash(), resp.peerID, resp.time)
+}
+
+// blockNoResponse message received from a peer
+type bcNoBlockResponse struct {
+	priorityNormal
+	time   time.Time
+	peerID p2p.ID
+	height int64
+}
+
+func (resp bcNoBlockResponse) String() string {
+	return fmt.Sprintf("bcNoBlockResponse{%v has no block at height %d at %v}",
+		resp.peerID, resp.height, resp.time)
+}
+
+// statusResponse message received from a peer
+type bcStatusResponse struct {
+	priorityNormal
+	time   time.Time
+	peerID p2p.ID
+	base   int64
+	height int64
+}
+
+func (resp bcStatusResponse) String() string {
+	return fmt.Sprintf("bcStatusResponse{%v is at height %d (base: %d) at %v}",
+		resp.peerID, resp.height, resp.base, resp.time)
+}
+
+// new peer is connected
+type bcAddNewPeer struct {
+	priorityNormal
+	peerID p2p.ID
+}
+
+func (resp bcAddNewPeer) String() string {
+	return fmt.Sprintf("bcAddNewPeer{%v}", resp.peerID)
+}
+
+// existing peer is removed
+type bcRemovePeer struct {
+	priorityHigh
+	peerID p2p.ID
+	reason interface{}
+}
+
+func (resp bcRemovePeer) String() string {
+	return fmt.Sprintf("bcRemovePeer{%v due to %v}", resp.peerID, resp.reason)
+}
+
+// resets the scheduler and processor state, e.g. following a switch from state syncing
+type bcResetState struct {
+	priorityHigh
+	state state.State
+}
+
+func (e bcResetState) String() string {
+	return fmt.Sprintf("bcResetState{%v}", e.state)
+}
+
+// Takes the channel as a parameter to avoid race conditions on r.events.
+func (r *BlockchainReactor) demux(events <-chan Event) {
+	var lastRate = 0.0
+	var lastHundred = time.Now()
+
+	var (
+		processBlockFreq = 20 * time.Millisecond
+		doProcessBlockCh = make(chan struct{}, 1)
+		doProcessBlockTk = time.NewTicker(processBlockFreq)
+	)
+	defer doProcessBlockTk.Stop()
+
+	var (
+		prunePeerFreq = 1 * time.Second
+		doPrunePeerCh = make(chan struct{}, 1)
+		doPrunePeerTk = time.NewTicker(prunePeerFreq)
+	)
+	defer doPrunePeerTk.Stop()
+
+	var (
+		scheduleFreq = 20 * time.Millisecond
+		doScheduleCh = make(chan struct{}, 1)
+		doScheduleTk = time.NewTicker(scheduleFreq)
+	)
+	defer doScheduleTk.Stop()
+
+	var (
+		statusFreq = 10 * time.Second
+		doStatusCh = make(chan struct{}, 1)
+		doStatusTk = time.NewTicker(statusFreq)
+	)
+	defer doStatusTk.Stop()
+	doStatusCh <- struct{}{} // immediately broadcast to get status of existing peers
+
+	// XXX: Extract timers to make testing atemporal
+	for {
+		select {
+		// Pacers: send at most per frequency but don't saturate
+		case <-doProcessBlockTk.C:
+			select {
+			case doProcessBlockCh <- struct{}{}:
+			default:
+			}
+		case <-doPrunePeerTk.C:
+			select {
+			case doPrunePeerCh <- struct{}{}:
+			default:
+			}
+		case <-doScheduleTk.C:
+			select {
+			case doScheduleCh <- struct{}{}:
+			default:
+			}
+		case <-doStatusTk.C:
+			select {
+			case doStatusCh <- struct{}{}:
+			default:
+			}
+
+		// Tickers: perform tasks periodically
+		case <-doScheduleCh:
+			r.scheduler.send(rTrySchedule{time: time.Now()})
+		case <-doPrunePeerCh:
+			r.scheduler.send(rTryPrunePeer{time: time.Now()})
+		case <-doProcessBlockCh:
+			r.processor.send(rProcessBlock{})
+		case <-doStatusCh:
+			r.io.broadcastStatusRequest()
+
+		// Events from peers. Closing the channel signals event loop termination.
+		case event, ok := <-events:
+			if !ok {
+				r.logger.Info("Stopping event processing")
+				return
+			}
+			switch event := event.(type) {
+			case bcStatusResponse:
+				r.setMaxPeerHeight(event.height)
+				r.scheduler.send(event)
+			case bcAddNewPeer, bcRemovePeer, bcBlockResponse, bcNoBlockResponse:
+				r.scheduler.send(event)
+			default:
+				r.logger.Error("Received unexpected event", "event", fmt.Sprintf("%T", event))
+			}
+
+		// Incremental events from scheduler
+		case event := <-r.scheduler.next():
+			switch event := event.(type) {
+			case scBlockReceived:
+				r.processor.send(event)
+			case scPeerError:
+				r.processor.send(event)
+				if err := r.reporter.Report(behaviour.BadMessage(event.peerID, "scPeerError")); err != nil {
+					r.logger.Error("Error reporting peer", "err", err)
+				}
+			case scBlockRequest:
+				if err := r.io.sendBlockRequest(event.peerID, event.height); err != nil {
+					r.logger.Error("Error sending block request", "err", err)
+				}
+			case scFinishedEv:
+				r.processor.send(event)
+				r.scheduler.stop()
+			case scSchedulerFail:
+				r.logger.Error("Scheduler failure", "err", event.reason.Error())
+			case scPeersPruned:
+				// Remove peers from the processor.
+				for _, peerID := range event.peers {
+					r.processor.send(scPeerError{peerID: peerID, reason: errors.New("peer was pruned")})
+				}
+				r.logger.Debug("Pruned peers", "count", len(event.peers))
+			case noOpEvent:
+			default:
+				r.logger.Error("Received unexpected scheduler event", "event", fmt.Sprintf("%T", event))
+			}
+
+		// Incremental events from processor
+		case event := <-r.processor.next():
+			switch event := event.(type) {
+			case pcBlockProcessed:
+				r.setSyncHeight(event.height)
+				if r.syncHeight%100 == 0 {
+					lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
+					r.logger.Info("Fast Sync Rate", "height", r.syncHeight,
+						"max_peer_height", r.maxPeerHeight, "blocks/s", lastRate)
+					lastHundred = time.Now()
+				}
+				r.scheduler.send(event)
+			case pcBlockVerificationFailure:
+				r.scheduler.send(event)
+			case pcFinished:
+				r.logger.Info("Fast sync complete, switching to consensus")
+				if !r.io.trySwitchToConsensus(event.tmState, event.blocksSynced > 0 || r.stateSynced) {
+					r.logger.Error("Failed to switch to consensus reactor")
+				}
+				r.endSync()
+				return
+			case noOpEvent:
+			default:
+				r.logger.Error("Received unexpected processor event", "event", fmt.Sprintf("%T", event))
+			}
+
+		// Terminal event from scheduler
+		case err := <-r.scheduler.final():
+			switch err {
+			case nil:
+				r.logger.Info("Scheduler stopped")
+			default:
+				r.logger.Error("Scheduler aborted with error", "err", err)
+			}
+
+		// Terminal event from processor
+		case err := <-r.processor.final():
+			switch err {
+			case nil:
+				r.logger.Info("Processor stopped")
+			default:
+				r.logger.Error("Processor aborted with error", "err", err)
+			}
+		}
+	}
+}
+
+// Stop implements cmn.Service interface.
+func (r *BlockchainReactor) Stop() error {
+	r.logger.Info("reactor stopping")
+	r.endSync()
+	r.logger.Info("reactor stopped")
+	return nil
+}
+
+// Receive implements Reactor by handling different message types.
+func (r *BlockchainReactor) NewReceive(e p2p.Envelope) {
+	if err := bc.ValidateMsg(e.Message); err != nil {
+		r.logger.Error("peer sent us invalid msg", "peer", e.Src, "msg", e.Message, "err", err)
+		_ = r.reporter.Report(behaviour.BadMessage(e.Src.ID(), err.Error()))
+		return
+	}
+
+	r.logger.Debug("Receive", "src", e.Src.ID(), "chID", e.ChannelID, "msg", e.Message)
+
+	switch msg := e.Message.(type) {
+	case *bcproto.StatusRequest:
+		if err := r.io.sendStatusResponse(r.store.Base(), r.store.Height(), e.Src.ID()); err != nil {
+			r.logger.Error("Could not send status message to peer", "src", e.Src)
+		}
+
+	case *bcproto.BlockRequest:
+		block := r.store.LoadBlock(msg.Height)
+		if block != nil {
+			if err := r.io.sendBlockToPeer(block, e.Src.ID()); err != nil {
+				r.logger.Error("Could not send block message to peer: ", err)
+			}
+		} else {
+			r.logger.Info("peer asking for a block we don't have", "src", e.Src, "height", msg.Height)
+			peerID := e.Src.ID()
+			if err := r.io.sendBlockNotFound(msg.Height, peerID); err != nil {
+				r.logger.Error("Couldn't send block not found: ", err)
+			}
+		}
+
+	case *bcproto.StatusResponse:
+		r.mtx.RLock()
+		if r.events != nil {
+			r.events <- bcStatusResponse{peerID: e.Src.ID(), base: msg.Base, height: msg.Height}
+		}
+		r.mtx.RUnlock()
+
+	case *bcproto.BlockResponse:
+		bi, err := types.BlockFromProto(msg.Block)
+		if err != nil {
+			r.logger.Error("error transitioning block from protobuf", "err", err)
+			return
+		}
+		r.mtx.RLock()
+		if r.events != nil {
+			r.events <- bcBlockResponse{
+				peerID: e.Src.ID(),
+				block:  bi,
+				time:   time.Now(),
+			}
+		}
+		r.mtx.RUnlock()
+
+	case *bcproto.NoBlockResponse:
+		r.mtx.RLock()
+		if r.events != nil {
+			r.events <- bcNoBlockResponse{peerID: e.Src.ID(), height: msg.Height, time: time.Now()}
+		}
+		r.mtx.RUnlock()
+	}
+}
+
+func (r *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
+// AddPeer implements Reactor interface
+func (r *BlockchainReactor) AddPeer(peer p2p.Peer) {
+	err := r.io.sendStatusResponse(r.store.Base(), r.store.Height(), peer.ID())
+	if err != nil {
+		r.logger.Error("Could not send status message to peer new", "src", peer.ID, "height", r.SyncHeight())
+	}
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+	if r.events != nil {
+		r.events <- bcAddNewPeer{peerID: peer.ID()}
+	}
+}
+
+// RemovePeer implements Reactor interface.
+func (r *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+	if r.events != nil {
+		r.events <- bcRemovePeer{
+			peerID: peer.ID(),
+			reason: reason,
+		}
+	}
+}
+
+// GetChannels implements Reactor
+func (r *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
+	return []*p2p.ChannelDescriptor{
+		{
+			ID:                  BlockchainChannel,
+			Priority:            5,
+			SendQueueCapacity:   2000,
+			RecvBufferCapacity:  50 * 4096,
+			RecvMessageCapacity: bc.MaxMsgSize,
+		},
+	}
+}

blocksync/reactor.go

@@ -233,6 +233,8 @@ func (bcR *Reactor) NewReceive(e p2p.Envelope) {
 	}
 }
 
+func (bcR *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // Handle messages from the poolReactor telling the reactor what to do.
 // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
 func (bcR *Reactor) poolRoutine(stateSynced bool) {

consensus/reactor.go

@@ -387,6 +387,8 @@ func (conR *Reactor) NewReceive(e p2p.Envelope) {
 	}
 }
 
+func (conR *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // SetEventBus sets event bus.
 func (conR *Reactor) SetEventBus(b *types.EventBus) {
 	conR.eventBus = b

evidence/reactor.go

@@ -92,6 +92,8 @@ func (evR *Reactor) NewReceive(e p2p.Envelope) {
 	}
 }
 
+func (evR *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // SetEventBus implements events.Eventable.
 func (evR *Reactor) SetEventBus(b *types.EventBus) {
 	evR.eventBus = b

mempool/v1/reactor.go

@@ -187,6 +187,8 @@ func (memR *Reactor) NewReceive(e p2p.Envelope) {
 	// broadcasting happens from go routines per peer
 }
 
+func (memR *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // PeerState describes the state of a peer.
 type PeerState interface {
 	GetHeight() int64

p2p/base_reactor.go

@@ -82,5 +82,7 @@ func (*BaseReactor) GetChannels() []*conn.ChannelDescriptor        { return nil
 func (*BaseReactor) AddPeer(peer Peer)                             {}
 func (*BaseReactor) RemovePeer(peer Peer, reason interface{})      {}
 func (*BaseReactor) NewReceive(e Envelope)                         {}
+
+func (*BaseReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte)                         {}}
 func (*BaseReactor) Receive(chID byte, peer Peer, msgBytes []byte) {}
 func (*BaseReactor) InitPeer(peer Peer) Peer                       { return peer }

p2p/mock/reactor.go

@@ -23,3 +23,5 @@ func (r *Reactor) GetChannels() []*conn.ChannelDescriptor       { return r.Chann
 func (r *Reactor) AddPeer(peer p2p.Peer)                        {}
 func (r *Reactor) RemovePeer(peer p2p.Peer, reason interface{}) {}
 func (r *Reactor) NewReceive(e p2p.Envelope)                    {}
+
+func (r *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte)                    {}}

p2p/pex/pex_reactor.go

@@ -298,6 +298,8 @@ func (r *Reactor) NewReceive(e p2p.Envelope) {
 	}
 }
 
+func (r *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // enforces a minimum amount of time between requests
 func (r *Reactor) receiveRequest(src Peer) error {
 	id := string(src.ID())

p2p/switch_test.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/cosmos/gogoproto/proto"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -81,6 +82,8 @@ func (tr *TestReactor) NewReceive(e Envelope) {
 	}
 }
 
+func (tr *TestReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 func (tr *TestReactor) getMsgs(chID byte) []PeerMessage {
 	tr.mtx.Lock()
 	defer tr.mtx.Unlock()

statesync/reactor.go

@@ -223,6 +223,8 @@ func (r *Reactor) NewReceive(e p2p.Envelope) {
 	}
 }
 
+func (r *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}
+
 // recentSnapshots fetches the n most recent snapshots from the app
 func (r *Reactor) recentSnapshots(n uint32) ([]*snapshot, error) {
 	resp, err := r.conn.ListSnapshotsSync(abci.RequestListSnapshots{})