allow BlockchainReactor to reset ConsensusReactor state

blockchain/pool.go

@@ -339,16 +339,19 @@ func requestRoutine(pool *BlockPool, height uint) {
 	PICK_LOOP:
 		for {
 			if !pool.IsRunning() {
+				log.Debug("BlockPool not running. Stopping requestRoutine", "height", height)
 				return
 			}
 			peer = pool.pickIncrAvailablePeer(height)
 			if peer == nil {
+				log.Debug("No peers available", "height", height)
 				time.Sleep(requestIntervalMS * time.Millisecond)
 				continue PICK_LOOP
 			}
 			break PICK_LOOP
 		}
 
+		log.Debug("Selected peer for request", "height", height, "peerId", peer.id)
 		pool.setPeerForRequest(height, peer.id)
 
 		for try := 0; try < maxTries; try++ {

blockchain/reactor.go

@@ -25,6 +25,10 @@ const (
 	stopSyncingDurationMinutes = 10
 )
 
+type stateResetter interface {
+	ResetToState(*sm.State)
+}
+
 // BlockchainReactor handles long-term catchup syncing.
 type BlockchainReactor struct {
 	sw         *p2p.Switch
@@ -93,6 +97,7 @@ func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
 
 // Implements Reactor
 func (bcR *BlockchainReactor) AddPeer(peer *p2p.Peer) {
+	log.Debug("BlockchainReactor AddPeer", "peer", peer)
 	// Send peer our state.
 	peer.Send(BlockchainChannel, PeerStatusMessage{bcR.store.Height()})
 }
@@ -203,6 +208,7 @@ FOR_LOOP:
 					//bcR.sw.Reactor("BLOCKCHAIN").Stop()
 					trySyncTicker.Stop() // Just stop the block requests.  Still serve blocks to others.
 					conR := bcR.sw.Reactor("CONSENSUS")
+					conR.(stateResetter).ResetToState(bcR.state)
 					conR.Start(bcR.sw)
 					for _, peer := range bcR.sw.Peers().List() {
 						conR.AddPeer(peer)

consensus/reactor.go

@@ -225,6 +225,7 @@ func (conR *ConsensusReactor) SetPrivValidator(priv *sm.PrivValidator) {
 	conR.conS.SetPrivValidator(priv)
 }
 
+// Fast forward to some state.
 func (conR *ConsensusReactor) UpdateToState(state *sm.State) {
 	conR.conS.updateToState(state, false)
 }