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.

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