p2p: state sync reactor refactor (#5671)

node/node.go

@@ -483,7 +483,7 @@ func createSwitch(config *cfg.Config,
 	peerFilters []p2p.PeerFilterFunc,
 	mempoolReactor *mempl.Reactor,
 	bcReactor p2p.Reactor,
-	stateSyncReactor *statesync.Reactor,
+	stateSyncReactor *p2p.ReactorShim,
 	consensusReactor *cs.Reactor,
 	evidenceReactor *evidence.Reactor,
 	nodeInfo p2p.NodeInfo,
@@ -746,9 +746,18 @@ func NewNode(config *cfg.Config,
 	// FIXME The way we do phased startups (e.g. replay -> fast sync -> consensus) is very messy,
 	// we should clean this whole thing up. See:
 	// https://github.com/tendermint/tendermint/issues/4644
-	stateSyncReactor := statesync.NewReactor(proxyApp.Snapshot(), proxyApp.Query(),
-		config.StateSync.TempDir)
-	stateSyncReactor.SetLogger(logger.With("module", "statesync"))
+	stateSyncReactorShim := p2p.NewReactorShim("StateSyncShim", statesync.ChannelShims)
+	stateSyncReactorShim.SetLogger(logger.With("module", "statesync"))
+
+	stateSyncReactor := statesync.NewReactor(
+		stateSyncReactorShim.Logger,
+		proxyApp.Snapshot(),
+		proxyApp.Query(),
+		stateSyncReactorShim.GetChannel(statesync.SnapshotChannel),
+		stateSyncReactorShim.GetChannel(statesync.ChunkChannel),
+		stateSyncReactorShim.PeerUpdates,
+		config.StateSync.TempDir,
+	)
 
 	nodeInfo, err := makeNodeInfo(config, nodeKey, txIndexer, genDoc, state)
 	if err != nil {
@@ -762,7 +771,7 @@ func NewNode(config *cfg.Config,
 	p2pLogger := logger.With("module", "p2p")
 	sw := createSwitch(
 		config, transport, p2pMetrics, peerFilters, mempoolReactor, bcReactor,
-		stateSyncReactor, consensusReactor, evidenceReactor, nodeInfo, nodeKey, p2pLogger,
+		stateSyncReactorShim, consensusReactor, evidenceReactor, nodeInfo, nodeKey, p2pLogger,
 	)
 
 	err = sw.AddPersistentPeers(splitAndTrimEmpty(config.P2P.PersistentPeers, ",", " "))
@@ -892,6 +901,11 @@ func (n *Node) OnStart() error {
 		return err
 	}
 
+	// Start the real state sync reactor separately since the switch uses the shim.
+	if err := n.stateSyncReactor.Start(); err != nil {
+		return err
+	}
+
 	// Always connect to persistent peers
 	err = n.sw.DialPeersAsync(splitAndTrimEmpty(n.config.P2P.PersistentPeers, ",", " "))
 	if err != nil {
@@ -933,6 +947,11 @@ func (n *Node) OnStop() {
 		n.Logger.Error("Error closing switch", "err", err)
 	}
 
+	// Stop the real state sync reactor separately since the switch uses the shim.
+	if err := n.stateSyncReactor.Stop(); err != nil {
+		n.Logger.Error("failed to stop state sync service", "err", err)
+	}
+
 	// stop mempool WAL
 	if n.config.Mempool.WalEnabled() {
 		n.mempool.CloseWAL()
@@ -1255,7 +1274,7 @@ func makeNodeInfo(
 			cs.StateChannel, cs.DataChannel, cs.VoteChannel, cs.VoteSetBitsChannel,
 			mempl.MempoolChannel,
 			evidence.EvidenceChannel,
-			statesync.SnapshotChannel, statesync.ChunkChannel,
+			byte(statesync.SnapshotChannel), byte(statesync.ChunkChannel),
 		},
 		Moniker: config.Moniker,
 		Other: p2p.DefaultNodeInfoOther{