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.

822 lines
23 KiB

  1. package statesync
  2. import (
  3. "bytes"
  4. "context"
  5. "errors"
  6. "fmt"
  7. "reflect"
  8. "sort"
  9. "time"
  10. abci "github.com/tendermint/tendermint/abci/types"
  11. "github.com/tendermint/tendermint/config"
  12. tmsync "github.com/tendermint/tendermint/internal/libs/sync"
  13. "github.com/tendermint/tendermint/internal/p2p"
  14. "github.com/tendermint/tendermint/libs/log"
  15. "github.com/tendermint/tendermint/libs/service"
  16. ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
  17. "github.com/tendermint/tendermint/proxy"
  18. sm "github.com/tendermint/tendermint/state"
  19. "github.com/tendermint/tendermint/store"
  20. "github.com/tendermint/tendermint/types"
  21. )
  22. var (
  23. _ service.Service = (*Reactor)(nil)
  24. _ p2p.Wrapper = (*ssproto.Message)(nil)
  25. // ChannelShims contains a map of ChannelDescriptorShim objects, where each
  26. // object wraps a reference to a legacy p2p ChannelDescriptor and the corresponding
  27. // p2p proto.Message the new p2p Channel is responsible for handling.
  28. //
  29. //
  30. // TODO: Remove once p2p refactor is complete.
  31. // ref: https://github.com/tendermint/tendermint/issues/5670
  32. ChannelShims = map[p2p.ChannelID]*p2p.ChannelDescriptorShim{
  33. SnapshotChannel: {
  34. MsgType: new(ssproto.Message),
  35. Descriptor: &p2p.ChannelDescriptor{
  36. ID: byte(SnapshotChannel),
  37. Priority: 5,
  38. SendQueueCapacity: 10,
  39. RecvMessageCapacity: snapshotMsgSize,
  40. MaxSendBytes: 400,
  41. },
  42. },
  43. ChunkChannel: {
  44. MsgType: new(ssproto.Message),
  45. Descriptor: &p2p.ChannelDescriptor{
  46. ID: byte(ChunkChannel),
  47. Priority: 1,
  48. SendQueueCapacity: 4,
  49. RecvMessageCapacity: chunkMsgSize,
  50. MaxSendBytes: 400,
  51. },
  52. },
  53. LightBlockChannel: {
  54. MsgType: new(ssproto.Message),
  55. Descriptor: &p2p.ChannelDescriptor{
  56. ID: byte(LightBlockChannel),
  57. Priority: 1,
  58. SendQueueCapacity: 10,
  59. RecvMessageCapacity: lightBlockMsgSize,
  60. MaxSendBytes: 400,
  61. },
  62. },
  63. }
  64. )
  65. const (
  66. // SnapshotChannel exchanges snapshot metadata
  67. SnapshotChannel = p2p.ChannelID(0x60)
  68. // ChunkChannel exchanges chunk contents
  69. ChunkChannel = p2p.ChannelID(0x61)
  70. // LightBlockChannel exchanges light blocks
  71. LightBlockChannel = p2p.ChannelID(0x62)
  72. // recentSnapshots is the number of recent snapshots to send and receive per peer.
  73. recentSnapshots = 10
  74. // snapshotMsgSize is the maximum size of a snapshotResponseMessage
  75. snapshotMsgSize = int(4e6) // ~4MB
  76. // chunkMsgSize is the maximum size of a chunkResponseMessage
  77. chunkMsgSize = int(16e6) // ~16MB
  78. // lightBlockMsgSize is the maximum size of a lightBlockResponseMessage
  79. lightBlockMsgSize = int(1e7) // ~10MB
  80. // lightBlockResponseTimeout is how long the dispatcher waits for a peer to
  81. // return a light block
  82. lightBlockResponseTimeout = 10 * time.Second
  83. // maxLightBlockRequestRetries is the amount of retries acceptable before
  84. // the backfill process aborts
  85. maxLightBlockRequestRetries = 20
  86. // the amount of processes fetching light blocks - this should be roughly calculated
  87. // as the time to fetch a block / time to verify a block
  88. lightBlockFetchers = 4
  89. )
  90. // Reactor handles state sync, both restoring snapshots for the local node and
  91. // serving snapshots for other nodes.
  92. type Reactor struct {
  93. service.BaseService
  94. cfg config.StateSyncConfig
  95. stateStore sm.Store
  96. blockStore *store.BlockStore
  97. conn proxy.AppConnSnapshot
  98. connQuery proxy.AppConnQuery
  99. tempDir string
  100. snapshotCh *p2p.Channel
  101. chunkCh *p2p.Channel
  102. blockCh *p2p.Channel
  103. peerUpdates *p2p.PeerUpdates
  104. closeCh chan struct{}
  105. dispatcher *dispatcher
  106. // This will only be set when a state sync is in progress. It is used to feed
  107. // received snapshots and chunks into the sync.
  108. mtx tmsync.RWMutex
  109. syncer *syncer
  110. }
  111. // NewReactor returns a reference to a new state sync reactor, which implements
  112. // the service.Service interface. It accepts a logger, connections for snapshots
  113. // and querying, references to p2p Channels and a channel to listen for peer
  114. // updates on. Note, the reactor will close all p2p Channels when stopping.
  115. func NewReactor(
  116. cfg config.StateSyncConfig,
  117. logger log.Logger,
  118. conn proxy.AppConnSnapshot,
  119. connQuery proxy.AppConnQuery,
  120. snapshotCh, chunkCh, blockCh *p2p.Channel,
  121. peerUpdates *p2p.PeerUpdates,
  122. stateStore sm.Store,
  123. blockStore *store.BlockStore,
  124. tempDir string,
  125. ) *Reactor {
  126. r := &Reactor{
  127. cfg: cfg,
  128. conn: conn,
  129. connQuery: connQuery,
  130. snapshotCh: snapshotCh,
  131. chunkCh: chunkCh,
  132. blockCh: blockCh,
  133. peerUpdates: peerUpdates,
  134. closeCh: make(chan struct{}),
  135. tempDir: tempDir,
  136. dispatcher: newDispatcher(blockCh.Out, lightBlockResponseTimeout),
  137. stateStore: stateStore,
  138. blockStore: blockStore,
  139. }
  140. r.BaseService = *service.NewBaseService(logger, "StateSync", r)
  141. return r
  142. }
  143. // OnStart starts separate go routines for each p2p Channel and listens for
  144. // envelopes on each. In addition, it also listens for peer updates and handles
  145. // messages on that p2p channel accordingly. The caller must be sure to execute
  146. // OnStop to ensure the outbound p2p Channels are closed. No error is returned.
  147. func (r *Reactor) OnStart() error {
  148. // Listen for envelopes on the snapshot p2p Channel in a separate go-routine
  149. // as to not block or cause IO contention with the chunk p2p Channel. Note,
  150. // we do not launch a go-routine to handle individual envelopes as to not
  151. // have to deal with bounding workers or pools.
  152. go r.processSnapshotCh()
  153. // Listen for envelopes on the chunk p2p Channel in a separate go-routine
  154. // as to not block or cause IO contention with the snapshot p2p Channel. Note,
  155. // we do not launch a go-routine to handle individual envelopes as to not
  156. // have to deal with bounding workers or pools.
  157. go r.processChunkCh()
  158. go r.processBlockCh()
  159. go r.processPeerUpdates()
  160. r.dispatcher.start()
  161. return nil
  162. }
  163. // OnStop stops the reactor by signaling to all spawned goroutines to exit and
  164. // blocking until they all exit.
  165. func (r *Reactor) OnStop() {
  166. // tell the dispatcher to stop sending any more requests
  167. r.dispatcher.stop()
  168. // Close closeCh to signal to all spawned goroutines to gracefully exit. All
  169. // p2p Channels should execute Close().
  170. close(r.closeCh)
  171. // Wait for all p2p Channels to be closed before returning. This ensures we
  172. // can easily reason about synchronization of all p2p Channels and ensure no
  173. // panics will occur.
  174. <-r.snapshotCh.Done()
  175. <-r.chunkCh.Done()
  176. <-r.blockCh.Done()
  177. <-r.peerUpdates.Done()
  178. }
  179. // Sync runs a state sync, fetching snapshots and providing chunks to the
  180. // application. It also saves tendermint state and runs a backfill process to
  181. // retrieve the necessary amount of headers, commits and validators sets to be
  182. // able to process evidence and participate in consensus.
  183. func (r *Reactor) Sync(stateProvider StateProvider, discoveryTime time.Duration) (sm.State, error) {
  184. r.mtx.Lock()
  185. if r.syncer != nil {
  186. r.mtx.Unlock()
  187. return sm.State{}, errors.New("a state sync is already in progress")
  188. }
  189. r.syncer = newSyncer(
  190. r.cfg,
  191. r.Logger,
  192. r.conn,
  193. r.connQuery,
  194. stateProvider,
  195. r.snapshotCh.Out,
  196. r.chunkCh.Out,
  197. r.tempDir,
  198. )
  199. r.mtx.Unlock()
  200. hook := func() {
  201. // request snapshots from all currently connected peers
  202. r.Logger.Debug("requesting snapshots from known peers")
  203. r.snapshotCh.Out <- p2p.Envelope{
  204. Broadcast: true,
  205. Message: &ssproto.SnapshotsRequest{},
  206. }
  207. }
  208. hook()
  209. state, commit, err := r.syncer.SyncAny(discoveryTime, hook)
  210. if err != nil {
  211. return sm.State{}, err
  212. }
  213. r.mtx.Lock()
  214. r.syncer = nil
  215. r.mtx.Unlock()
  216. err = r.stateStore.Bootstrap(state)
  217. if err != nil {
  218. return sm.State{}, fmt.Errorf("failed to bootstrap node with new state: %w", err)
  219. }
  220. err = r.blockStore.SaveSeenCommit(state.LastBlockHeight, commit)
  221. if err != nil {
  222. return sm.State{}, fmt.Errorf("failed to store last seen commit: %w", err)
  223. }
  224. return state, nil
  225. }
  226. // Backfill sequentially fetches, verifies and stores light blocks in reverse
  227. // order. It does not stop verifying blocks until reaching a block with a height
  228. // and time that is less or equal to the stopHeight and stopTime. The
  229. // trustedBlockID should be of the header at startHeight.
  230. func (r *Reactor) Backfill(state sm.State) error {
  231. params := state.ConsensusParams.Evidence
  232. stopHeight := state.LastBlockHeight - params.MaxAgeNumBlocks
  233. stopTime := state.LastBlockTime.Add(-params.MaxAgeDuration)
  234. // ensure that stop height doesn't go below the initial height
  235. if stopHeight < state.InitialHeight {
  236. stopHeight = state.InitialHeight
  237. // this essentially makes stop time a void criteria for termination
  238. stopTime = state.LastBlockTime
  239. }
  240. return r.backfill(
  241. context.Background(),
  242. state.ChainID,
  243. state.LastBlockHeight, stopHeight,
  244. state.LastBlockID,
  245. stopTime,
  246. )
  247. }
  248. func (r *Reactor) backfill(
  249. ctx context.Context,
  250. chainID string,
  251. startHeight, stopHeight int64,
  252. trustedBlockID types.BlockID,
  253. stopTime time.Time,
  254. ) error {
  255. r.Logger.Info("starting backfill process...", "startHeight", startHeight,
  256. "stopHeight", stopHeight, "trustedBlockID", trustedBlockID)
  257. const sleepTime = 1 * time.Second
  258. var (
  259. lastValidatorSet *types.ValidatorSet
  260. lastChangeHeight int64 = startHeight
  261. )
  262. queue := newBlockQueue(startHeight, stopHeight, stopTime, maxLightBlockRequestRetries)
  263. // fetch light blocks across four workers. The aim with deploying concurrent
  264. // workers is to equate the network messaging time with the verification
  265. // time. Ideally we want the verification process to never have to be
  266. // waiting on blocks. If it takes 4s to retrieve a block and 1s to verify
  267. // it, then steady state involves four workers.
  268. for i := 0; i < lightBlockFetchers; i++ {
  269. go func() {
  270. for {
  271. select {
  272. case height := <-queue.nextHeight():
  273. r.Logger.Debug("fetching next block", "height", height)
  274. lb, peer, err := r.dispatcher.LightBlock(ctx, height)
  275. if err != nil {
  276. queue.retry(height)
  277. if errors.Is(err, errNoConnectedPeers) {
  278. r.Logger.Info("backfill: no connected peers to fetch light blocks from; sleeping...",
  279. "sleepTime", sleepTime)
  280. time.Sleep(sleepTime)
  281. } else {
  282. // we don't punish the peer as it might just have not responded in time
  283. r.Logger.Info("backfill: error with fetching light block",
  284. "height", height, "err", err)
  285. }
  286. continue
  287. }
  288. if lb == nil {
  289. r.Logger.Info("backfill: peer didn't have block, fetching from another peer", "height", height)
  290. queue.retry(height)
  291. // as we are fetching blocks backwards, if this node doesn't have the block it likely doesn't
  292. // have any prior ones, thus we remove it from the peer list
  293. r.dispatcher.removePeer(peer)
  294. continue
  295. }
  296. // run a validate basic. This checks the validator set and commit
  297. // hashes line up
  298. err = lb.ValidateBasic(chainID)
  299. if err != nil || lb.Height != height {
  300. r.Logger.Info("backfill: fetched light block failed validate basic, removing peer...",
  301. "err", err, "height", height)
  302. queue.retry(height)
  303. r.blockCh.Error <- p2p.PeerError{
  304. NodeID: peer,
  305. Err: fmt.Errorf("received invalid light block: %w", err),
  306. }
  307. continue
  308. }
  309. // add block to queue to be verified
  310. queue.add(lightBlockResponse{
  311. block: lb,
  312. peer: peer,
  313. })
  314. r.Logger.Debug("backfill: added light block to processing queue", "height", height)
  315. case <-queue.done():
  316. return
  317. }
  318. }
  319. }()
  320. }
  321. // verify all light blocks
  322. for {
  323. select {
  324. case <-r.closeCh:
  325. queue.close()
  326. return nil
  327. case <-ctx.Done():
  328. queue.close()
  329. return nil
  330. case resp := <-queue.verifyNext():
  331. // validate the header hash. We take the last block id of the
  332. // previous header (i.e. one height above) as the trusted hash which
  333. // we equate to. ValidatorsHash and CommitHash have already been
  334. // checked in the `ValidateBasic`
  335. if w, g := trustedBlockID.Hash, resp.block.Hash(); !bytes.Equal(w, g) {
  336. r.Logger.Info("received invalid light block. header hash doesn't match trusted LastBlockID",
  337. "trustedHash", w, "receivedHash", g, "height", resp.block.Height)
  338. r.blockCh.Error <- p2p.PeerError{
  339. NodeID: resp.peer,
  340. Err: fmt.Errorf("received invalid light block. Expected hash %v, got: %v", w, g),
  341. }
  342. queue.retry(resp.block.Height)
  343. continue
  344. }
  345. // save the signed headers
  346. err := r.blockStore.SaveSignedHeader(resp.block.SignedHeader, trustedBlockID)
  347. if err != nil {
  348. return err
  349. }
  350. // check if there has been a change in the validator set
  351. if lastValidatorSet != nil && !bytes.Equal(resp.block.Header.ValidatorsHash, resp.block.Header.NextValidatorsHash) {
  352. // save all the heights that the last validator set was the same
  353. err = r.stateStore.SaveValidatorSets(resp.block.Height+1, lastChangeHeight, lastValidatorSet)
  354. if err != nil {
  355. return err
  356. }
  357. // update the lastChangeHeight
  358. lastChangeHeight = resp.block.Height
  359. }
  360. trustedBlockID = resp.block.LastBlockID
  361. queue.success(resp.block.Height)
  362. r.Logger.Info("backfill: verified and stored light block", "height", resp.block.Height)
  363. lastValidatorSet = resp.block.ValidatorSet
  364. case <-queue.done():
  365. if err := queue.error(); err != nil {
  366. return err
  367. }
  368. // save the final batch of validators
  369. if err := r.stateStore.SaveValidatorSets(queue.terminal.Height, lastChangeHeight, lastValidatorSet); err != nil {
  370. return err
  371. }
  372. r.Logger.Info("successfully completed backfill process", "endHeight", queue.terminal.Height)
  373. return nil
  374. }
  375. }
  376. }
  377. // Dispatcher exposes the dispatcher so that a state provider can use it for
  378. // light client verification
  379. func (r *Reactor) Dispatcher() *dispatcher { //nolint:golint
  380. return r.dispatcher
  381. }
  382. // handleSnapshotMessage handles envelopes sent from peers on the
  383. // SnapshotChannel. It returns an error only if the Envelope.Message is unknown
  384. // for this channel. This should never be called outside of handleMessage.
  385. func (r *Reactor) handleSnapshotMessage(envelope p2p.Envelope) error {
  386. logger := r.Logger.With("peer", envelope.From)
  387. switch msg := envelope.Message.(type) {
  388. case *ssproto.SnapshotsRequest:
  389. snapshots, err := r.recentSnapshots(recentSnapshots)
  390. if err != nil {
  391. logger.Error("failed to fetch snapshots", "err", err)
  392. return nil
  393. }
  394. for _, snapshot := range snapshots {
  395. logger.Debug(
  396. "advertising snapshot",
  397. "height", snapshot.Height,
  398. "format", snapshot.Format,
  399. )
  400. r.snapshotCh.Out <- p2p.Envelope{
  401. To: envelope.From,
  402. Message: &ssproto.SnapshotsResponse{
  403. Height: snapshot.Height,
  404. Format: snapshot.Format,
  405. Chunks: snapshot.Chunks,
  406. Hash: snapshot.Hash,
  407. Metadata: snapshot.Metadata,
  408. },
  409. }
  410. }
  411. case *ssproto.SnapshotsResponse:
  412. r.mtx.RLock()
  413. defer r.mtx.RUnlock()
  414. if r.syncer == nil {
  415. logger.Debug("received unexpected snapshot; no state sync in progress")
  416. return nil
  417. }
  418. logger.Debug("received snapshot", "height", msg.Height, "format", msg.Format)
  419. _, err := r.syncer.AddSnapshot(envelope.From, &snapshot{
  420. Height: msg.Height,
  421. Format: msg.Format,
  422. Chunks: msg.Chunks,
  423. Hash: msg.Hash,
  424. Metadata: msg.Metadata,
  425. })
  426. if err != nil {
  427. logger.Error(
  428. "failed to add snapshot",
  429. "height", msg.Height,
  430. "format", msg.Format,
  431. "err", err,
  432. "channel", r.snapshotCh.ID,
  433. )
  434. return nil
  435. }
  436. default:
  437. return fmt.Errorf("received unknown message: %T", msg)
  438. }
  439. return nil
  440. }
  441. // handleChunkMessage handles envelopes sent from peers on the ChunkChannel.
  442. // It returns an error only if the Envelope.Message is unknown for this channel.
  443. // This should never be called outside of handleMessage.
  444. func (r *Reactor) handleChunkMessage(envelope p2p.Envelope) error {
  445. switch msg := envelope.Message.(type) {
  446. case *ssproto.ChunkRequest:
  447. r.Logger.Debug(
  448. "received chunk request",
  449. "height", msg.Height,
  450. "format", msg.Format,
  451. "chunk", msg.Index,
  452. "peer", envelope.From,
  453. )
  454. resp, err := r.conn.LoadSnapshotChunkSync(context.Background(), abci.RequestLoadSnapshotChunk{
  455. Height: msg.Height,
  456. Format: msg.Format,
  457. Chunk: msg.Index,
  458. })
  459. if err != nil {
  460. r.Logger.Error(
  461. "failed to load chunk",
  462. "height", msg.Height,
  463. "format", msg.Format,
  464. "chunk", msg.Index,
  465. "err", err,
  466. "peer", envelope.From,
  467. )
  468. return nil
  469. }
  470. r.Logger.Debug(
  471. "sending chunk",
  472. "height", msg.Height,
  473. "format", msg.Format,
  474. "chunk", msg.Index,
  475. "peer", envelope.From,
  476. )
  477. r.chunkCh.Out <- p2p.Envelope{
  478. To: envelope.From,
  479. Message: &ssproto.ChunkResponse{
  480. Height: msg.Height,
  481. Format: msg.Format,
  482. Index: msg.Index,
  483. Chunk: resp.Chunk,
  484. Missing: resp.Chunk == nil,
  485. },
  486. }
  487. case *ssproto.ChunkResponse:
  488. r.mtx.RLock()
  489. defer r.mtx.RUnlock()
  490. if r.syncer == nil {
  491. r.Logger.Debug("received unexpected chunk; no state sync in progress", "peer", envelope.From)
  492. return nil
  493. }
  494. r.Logger.Debug(
  495. "received chunk; adding to sync",
  496. "height", msg.Height,
  497. "format", msg.Format,
  498. "chunk", msg.Index,
  499. "peer", envelope.From,
  500. )
  501. _, err := r.syncer.AddChunk(&chunk{
  502. Height: msg.Height,
  503. Format: msg.Format,
  504. Index: msg.Index,
  505. Chunk: msg.Chunk,
  506. Sender: envelope.From,
  507. })
  508. if err != nil {
  509. r.Logger.Error(
  510. "failed to add chunk",
  511. "height", msg.Height,
  512. "format", msg.Format,
  513. "chunk", msg.Index,
  514. "err", err,
  515. "peer", envelope.From,
  516. )
  517. return nil
  518. }
  519. default:
  520. return fmt.Errorf("received unknown message: %T", msg)
  521. }
  522. return nil
  523. }
  524. func (r *Reactor) handleLightBlockMessage(envelope p2p.Envelope) error {
  525. switch msg := envelope.Message.(type) {
  526. case *ssproto.LightBlockRequest:
  527. r.Logger.Info("received light block request", "height", msg.Height)
  528. lb, err := r.fetchLightBlock(msg.Height)
  529. if err != nil {
  530. r.Logger.Error("failed to retrieve light block", "err", err, "height", msg.Height)
  531. return err
  532. }
  533. lbproto, err := lb.ToProto()
  534. if err != nil {
  535. r.Logger.Error("marshaling light block to proto", "err", err)
  536. return nil
  537. }
  538. // NOTE: If we don't have the light block we will send a nil light block
  539. // back to the requested node, indicating that we don't have it.
  540. r.blockCh.Out <- p2p.Envelope{
  541. To: envelope.From,
  542. Message: &ssproto.LightBlockResponse{
  543. LightBlock: lbproto,
  544. },
  545. }
  546. case *ssproto.LightBlockResponse:
  547. if err := r.dispatcher.respond(msg.LightBlock, envelope.From); err != nil {
  548. r.Logger.Error("error processing light block response", "err", err)
  549. return err
  550. }
  551. default:
  552. return fmt.Errorf("received unknown message: %T", msg)
  553. }
  554. return nil
  555. }
  556. // handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
  557. // It will handle errors and any possible panics gracefully. A caller can handle
  558. // any error returned by sending a PeerError on the respective channel.
  559. func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
  560. defer func() {
  561. if e := recover(); e != nil {
  562. err = fmt.Errorf("panic in processing message: %v", e)
  563. }
  564. }()
  565. r.Logger.Debug("received message", "message", reflect.TypeOf(envelope.Message), "peer", envelope.From)
  566. switch chID {
  567. case SnapshotChannel:
  568. err = r.handleSnapshotMessage(envelope)
  569. case ChunkChannel:
  570. err = r.handleChunkMessage(envelope)
  571. case LightBlockChannel:
  572. err = r.handleLightBlockMessage(envelope)
  573. default:
  574. err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
  575. }
  576. return err
  577. }
  578. // processSnapshotCh initiates a blocking process where we listen for and handle
  579. // envelopes on the SnapshotChannel.
  580. func (r *Reactor) processSnapshotCh() {
  581. r.processCh(r.snapshotCh, "snapshot")
  582. }
  583. // processChunkCh initiates a blocking process where we listen for and handle
  584. // envelopes on the ChunkChannel.
  585. func (r *Reactor) processChunkCh() {
  586. r.processCh(r.chunkCh, "chunk")
  587. }
  588. // processBlockCh initiates a blocking process where we listen for and handle
  589. // envelopes on the LightBlockChannel.
  590. func (r *Reactor) processBlockCh() {
  591. r.processCh(r.blockCh, "light block")
  592. }
  593. // processCh routes state sync messages to their respective handlers. Any error
  594. // encountered during message execution will result in a PeerError being sent on
  595. // the respective channel. When the reactor is stopped, we will catch the signal
  596. // and close the p2p Channel gracefully.
  597. func (r *Reactor) processCh(ch *p2p.Channel, chName string) {
  598. defer ch.Close()
  599. for {
  600. select {
  601. case envelope := <-ch.In:
  602. if err := r.handleMessage(ch.ID, envelope); err != nil {
  603. r.Logger.Error(fmt.Sprintf("failed to process %s message", chName),
  604. "ch_id", ch.ID, "envelope", envelope, "err", err)
  605. ch.Error <- p2p.PeerError{
  606. NodeID: envelope.From,
  607. Err: err,
  608. }
  609. }
  610. case <-r.closeCh:
  611. r.Logger.Debug(fmt.Sprintf("stopped listening on %s channel; closing...", chName))
  612. return
  613. }
  614. }
  615. }
  616. // processPeerUpdate processes a PeerUpdate, returning an error upon failing to
  617. // handle the PeerUpdate or if a panic is recovered.
  618. func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
  619. r.Logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
  620. r.mtx.RLock()
  621. defer r.mtx.RUnlock()
  622. switch peerUpdate.Status {
  623. case p2p.PeerStatusUp:
  624. if r.syncer != nil {
  625. r.syncer.AddPeer(peerUpdate.NodeID)
  626. }
  627. r.dispatcher.addPeer(peerUpdate.NodeID)
  628. case p2p.PeerStatusDown:
  629. if r.syncer != nil {
  630. r.syncer.RemovePeer(peerUpdate.NodeID)
  631. }
  632. r.dispatcher.removePeer(peerUpdate.NodeID)
  633. }
  634. }
  635. // processPeerUpdates initiates a blocking process where we listen for and handle
  636. // PeerUpdate messages. When the reactor is stopped, we will catch the signal and
  637. // close the p2p PeerUpdatesCh gracefully.
  638. func (r *Reactor) processPeerUpdates() {
  639. defer r.peerUpdates.Close()
  640. for {
  641. select {
  642. case peerUpdate := <-r.peerUpdates.Updates():
  643. r.processPeerUpdate(peerUpdate)
  644. case <-r.closeCh:
  645. r.Logger.Debug("stopped listening on peer updates channel; closing...")
  646. return
  647. }
  648. }
  649. }
  650. // recentSnapshots fetches the n most recent snapshots from the app
  651. func (r *Reactor) recentSnapshots(n uint32) ([]*snapshot, error) {
  652. resp, err := r.conn.ListSnapshotsSync(context.Background(), abci.RequestListSnapshots{})
  653. if err != nil {
  654. return nil, err
  655. }
  656. sort.Slice(resp.Snapshots, func(i, j int) bool {
  657. a := resp.Snapshots[i]
  658. b := resp.Snapshots[j]
  659. switch {
  660. case a.Height > b.Height:
  661. return true
  662. case a.Height == b.Height && a.Format > b.Format:
  663. return true
  664. default:
  665. return false
  666. }
  667. })
  668. snapshots := make([]*snapshot, 0, n)
  669. for i, s := range resp.Snapshots {
  670. if i >= recentSnapshots {
  671. break
  672. }
  673. snapshots = append(snapshots, &snapshot{
  674. Height: s.Height,
  675. Format: s.Format,
  676. Chunks: s.Chunks,
  677. Hash: s.Hash,
  678. Metadata: s.Metadata,
  679. })
  680. }
  681. return snapshots, nil
  682. }
  683. // fetchLightBlock works out whether the node has a light block at a particular
  684. // height and if so returns it so it can be gossiped to peers
  685. func (r *Reactor) fetchLightBlock(height uint64) (*types.LightBlock, error) {
  686. h := int64(height)
  687. blockMeta := r.blockStore.LoadBlockMeta(h)
  688. if blockMeta == nil {
  689. return nil, nil
  690. }
  691. commit := r.blockStore.LoadBlockCommit(h)
  692. if commit == nil {
  693. return nil, nil
  694. }
  695. vals, err := r.stateStore.LoadValidators(h)
  696. if err != nil {
  697. return nil, err
  698. }
  699. if vals == nil {
  700. return nil, nil
  701. }
  702. return &types.LightBlock{
  703. SignedHeader: &types.SignedHeader{
  704. Header: &blockMeta.Header,
  705. Commit: commit,
  706. },
  707. ValidatorSet: vals,
  708. }, nil
  709. }