mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-20 06:54:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin' into jasmina/4457_block_sync_verification

Browse Source
This commit is contained in:
Jasmina Malicevic
2022-04-07 14:42:57 +02:00
parent 09f522b249 14f41ac5e3
commit 07fac4ccc0
46 changed files with 830 additions and 751 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/blocksync/reactor.go
+123 -50
View File
@@ -12,6 +12,7 @@ import (
"github.com/tendermint/tendermint/internal/consensus"
"github.com/tendermint/tendermint/internal/eventbus"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/internal/p2p/conn"
sm "github.com/tendermint/tendermint/internal/state"
"github.com/tendermint/tendermint/internal/store"
"github.com/tendermint/tendermint/libs/log"
@@ -47,6 +48,7 @@ func GetChannelDescriptor() *p2p.ChannelDescriptor {
SendQueueCapacity: 1000,
RecvBufferCapacity: 1024,
RecvMessageCapacity: MaxMsgSize,
Name: "blockSync",
}
}
@@ -130,8 +132,8 @@ type Reactor struct {
consReactor consensusReactor
blockSync *atomicBool
blockSyncCh *p2p.Channel
peerUpdates *p2p.PeerUpdates
chCreator p2p.ChannelCreator
peerEvents p2p.PeerEventSubscriber
requestsCh <-chan BlockRequest
errorsCh <-chan peerError
@@ -146,39 +148,33 @@ type Reactor struct {
// NewReactor returns new reactor instance.
func NewReactor(
ctx context.Context,
logger log.Logger,
stateStore sm.Store,
blockExec *sm.BlockExecutor,
store *store.BlockStore,
consReactor consensusReactor,
channelCreator p2p.ChannelCreator,
peerUpdates *p2p.PeerUpdates,
peerEvents p2p.PeerEventSubscriber,
blockSync bool,
metrics *consensus.Metrics,
eventBus *eventbus.EventBus,
) (*Reactor, error) {
blockSyncCh, err := channelCreator(ctx, GetChannelDescriptor())
if err != nil {
return nil, err
}
) *Reactor {
r := &Reactor{
logger: logger,
stateStore: stateStore,
blockExec: blockExec,
store: store,
consReactor: consReactor,
blockSync: newAtomicBool(blockSync),
blockSyncCh: blockSyncCh,
peerUpdates: peerUpdates,
metrics: metrics,
eventBus: eventBus,
logger: logger,
stateStore: stateStore,
blockExec: blockExec,
store: store,
consReactor: consReactor,
blockSync: newAtomicBool(blockSync),
chCreator: channelCreator,
peerEvents: peerEvents,
metrics: metrics,
eventBus: eventBus,
lastTrustedBlock: nil,
}
r.BaseService = *service.NewBaseService(logger, "BlockSync", r)
return r, nil
return r
}
// OnStart starts separate go routines for each p2p Channel and listens for
@@ -189,6 +185,12 @@ func NewReactor(
// If blockSync is enabled, we also start the pool and the pool processing
// goroutine. If the pool fails to start, an error is returned.
func (r *Reactor) OnStart(ctx context.Context) error {
blockSyncCh, err := r.chCreator(ctx, GetChannelDescriptor())
if err != nil {
return err
}
r.chCreator = func(context.Context, *conn.ChannelDescriptor) (*p2p.Channel, error) { return blockSyncCh, nil }
state, err := r.stateStore.Load()
if err != nil {
return err
@@ -214,13 +216,13 @@ func (r *Reactor) OnStart(ctx context.Context) error {
if err := r.pool.Start(ctx); err != nil {
return err
}
go r.requestRoutine(ctx)
go r.requestRoutine(ctx, blockSyncCh)
go r.poolRoutine(ctx, false)
go r.poolRoutine(ctx, false, blockSyncCh)
}
go r.processBlockSyncCh(ctx)
go r.processPeerUpdates(ctx)
go r.processBlockSyncCh(ctx, blockSyncCh)
go r.processPeerUpdates(ctx, r.peerEvents(ctx), blockSyncCh)
return nil
}
@@ -235,12 +237,12 @@ func (r *Reactor) OnStop() {
// respondToPeer loads a block and sends it to the requesting peer, if we have it.
// Otherwise, we'll respond saying we do not have it.
func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest, peerID types.NodeID) error {
func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest, peerID types.NodeID, blockSyncCh *p2p.Channel) error {
block := r.store.LoadBlockProto(msg.Height)
if block != nil {
blockCommit := r.store.LoadBlockCommitProto(msg.Height)
if blockCommit != nil {
return r.blockSyncCh.Send(ctx, p2p.Envelope{
return blockSyncCh.Send(ctx, p2p.Envelope{
To: peerID,
Message: &bcproto.BlockResponse{Block: block, Commit: blockCommit},
})
@@ -249,16 +251,17 @@ func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest,
r.logger.Info("peer requesting a block we do not have", "peer", peerID, "height", msg.Height)
return r.blockSyncCh.Send(ctx, p2p.Envelope{
return blockSyncCh.Send(ctx, p2p.Envelope{
To: peerID,
Message: &bcproto.NoBlockResponse{Height: msg.Height},
})
}
<<<<<<< HEAD
// handleBlockSyncMessage handles envelopes sent from peers on the
// BlockSyncChannel. It returns an error only if the Envelope.Message is unknown
// for this channel. This should never be called outside of handleMessage.
func (r *Reactor) handleBlockSyncMessage(ctx context.Context, envelope *p2p.Envelope) error {
func (r *Reactor) handleBlockSyncMessage(ctx context.Context, envelope *p2p.Envelope, blockSyncCh *p2p.Channel) error {
logger := r.logger.With("peer", envelope.From)
switch msg := envelope.Message.(type) {
@@ -299,10 +302,12 @@ func (r *Reactor) handleBlockSyncMessage(ctx context.Context, envelope *p2p.Enve
return nil
}
=======
>>>>>>> origin
// handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
// It will handle errors and any possible panics gracefully. A caller can handle
// any error returned by sending a PeerError on the respective channel.
func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelope *p2p.Envelope) (err error) {
func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelope *p2p.Envelope, blockSyncCh *p2p.Channel) (err error) {
defer func() {
if e := recover(); e != nil {
err = fmt.Errorf("panic in processing message: %v", e)
@@ -318,7 +323,39 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
switch chID {
case BlockSyncChannel:
err = r.handleBlockSyncMessage(ctx, envelope)
switch msg := envelope.Message.(type) {
case *bcproto.BlockRequest:
return r.respondToPeer(ctx, msg, envelope.From, blockSyncCh)
case *bcproto.BlockResponse:
block, err := types.BlockFromProto(msg.Block)
if err != nil {
r.logger.Error("failed to convert block from proto",
"peer", envelope.From,
"err", err)
return err
}
r.pool.AddBlock(envelope.From, block, block.Size())
case *bcproto.StatusRequest:
return blockSyncCh.Send(ctx, p2p.Envelope{
To: envelope.From,
Message: &bcproto.StatusResponse{
Height: r.store.Height(),
Base: r.store.Base(),
},
})
case *bcproto.StatusResponse:
r.pool.SetPeerRange(envelope.From, msg.Base, msg.Height)
case *bcproto.NoBlockResponse:
r.logger.Debug("peer does not have the requested block",
"peer", envelope.From,
"height", msg.Height)
default:
return fmt.Errorf("received unknown message: %T", msg)
}
default:
err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
@@ -332,17 +369,17 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
// message execution will result in a PeerError being sent on the BlockSyncChannel.
// When the reactor is stopped, we will catch the signal and close the p2p Channel
// gracefully.
func (r *Reactor) processBlockSyncCh(ctx context.Context) {
iter := r.blockSyncCh.Receive(ctx)
func (r *Reactor) processBlockSyncCh(ctx context.Context, blockSyncCh *p2p.Channel) {
iter := blockSyncCh.Receive(ctx)
for iter.Next(ctx) {
envelope := iter.Envelope()
if err := r.handleMessage(ctx, r.blockSyncCh.ID, envelope); err != nil {
if err := r.handleMessage(ctx, blockSyncCh.ID, envelope, blockSyncCh); err != nil {
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return
}
r.logger.Error("failed to process message", "ch_id", r.blockSyncCh.ID, "envelope", envelope, "err", err)
if serr := r.blockSyncCh.SendError(ctx, p2p.PeerError{
r.logger.Error("failed to process message", "ch_id", blockSyncCh.ID, "envelope", envelope, "err", err)
if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: envelope.From,
Err: err,
}); serr != nil {
@@ -353,7 +390,7 @@ func (r *Reactor) processBlockSyncCh(ctx context.Context) {
}
// processPeerUpdate processes a PeerUpdate.
func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpdate) {
func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpdate, blockSyncCh *p2p.Channel) {
r.logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
// XXX: Pool#RedoRequest can sometimes give us an empty peer.
@@ -364,7 +401,7 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
switch peerUpdate.Status {
case p2p.PeerStatusUp:
// send a status update the newly added peer
if err := r.blockSyncCh.Send(ctx, p2p.Envelope{
if err := blockSyncCh.Send(ctx, p2p.Envelope{
To: peerUpdate.NodeID,
Message: &bcproto.StatusResponse{
Base: r.store.Base(),
@@ -372,7 +409,7 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
},
}); err != nil {
r.pool.RemovePeer(peerUpdate.NodeID)
if err := r.blockSyncCh.SendError(ctx, p2p.PeerError{
if err := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: peerUpdate.NodeID,
Err: err,
}); err != nil {
@@ -388,13 +425,13 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
// processPeerUpdates initiates a blocking process where we listen for and handle
// PeerUpdate messages. When the reactor is stopped, we will catch the signal and
// close the p2p PeerUpdatesCh gracefully.
func (r *Reactor) processPeerUpdates(ctx context.Context) {
func (r *Reactor) processPeerUpdates(ctx context.Context, peerUpdates *p2p.PeerUpdates, blockSyncCh *p2p.Channel) {
for {
select {
case <-ctx.Done():
return
case peerUpdate := <-r.peerUpdates.Updates():
r.processPeerUpdate(ctx, peerUpdate)
case peerUpdate := <-peerUpdates.Updates():
r.processPeerUpdate(ctx, peerUpdate, blockSyncCh)
}
}
}
@@ -412,13 +449,18 @@ func (r *Reactor) SwitchToBlockSync(ctx context.Context, state sm.State) error {
r.syncStartTime = time.Now()
go r.requestRoutine(ctx)
go r.poolRoutine(ctx, true)
bsCh, err := r.chCreator(ctx, GetChannelDescriptor())
if err != nil {
return err
}
go r.requestRoutine(ctx, bsCh)
go r.poolRoutine(ctx, true, bsCh)
return nil
}
func (r *Reactor) requestRoutine(ctx context.Context) {
func (r *Reactor) requestRoutine(ctx context.Context, blockSyncCh *p2p.Channel) {
statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
defer statusUpdateTicker.Stop()
@@ -427,11 +469,11 @@ func (r *Reactor) requestRoutine(ctx context.Context) {
case <-ctx.Done():
return
case request := <-r.requestsCh:
if err := r.blockSyncCh.Send(ctx, p2p.Envelope{
if err := blockSyncCh.Send(ctx, p2p.Envelope{
To: request.PeerID,
Message: &bcproto.BlockRequest{Height: request.Height},
}); err != nil {
if err := r.blockSyncCh.SendError(ctx, p2p.PeerError{
if err := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: request.PeerID,
Err: err,
}); err != nil {
@@ -439,14 +481,14 @@ func (r *Reactor) requestRoutine(ctx context.Context) {
}
}
case pErr := <-r.errorsCh:
if err := r.blockSyncCh.SendError(ctx, p2p.PeerError{
if err := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: pErr.peerID,
Err: pErr.err,
}); err != nil {
return
}
case <-statusUpdateTicker.C:
if err := r.blockSyncCh.Send(ctx, p2p.Envelope{
if err := blockSyncCh.Send(ctx, p2p.Envelope{
Broadcast: true,
Message: &bcproto.StatusRequest{},
}); err != nil {
@@ -460,7 +502,7 @@ func (r *Reactor) requestRoutine(ctx context.Context) {
// do.
//
// NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool) {
func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh *p2p.Channel) {
var (
trySyncTicker = time.NewTicker(trySyncIntervalMS * time.Millisecond)
switchToConsensusTicker = time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
@@ -563,6 +605,7 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool) {
newBlockID = types.BlockID{Hash: newBlock.block.Hash(), PartSetHeader: newBlockPartSetHeader}
)
<<<<<<< HEAD
if r.lastTrustedBlock != nil && r.lastTrustedBlock.block != nil {
if newBlock.block.Height != r.lastTrustedBlock.block.Height+1 {
panic("Need last block")
@@ -587,6 +630,36 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool) {
peerID := r.pool.RedoRequest(r.lastTrustedBlock.block.Height + 1)
if serr := r.blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: peerID,
=======
// 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.
if err = state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit); err != nil {
err = fmt.Errorf("invalid last commit: %w", err)
r.logger.Error(
err.Error(),
"last_commit", second.LastCommit,
"block_id", firstID,
"height", first.Height,
)
// NOTE: We've already removed the peer's request, but we still need
// to clean up the rest.
peerID := r.pool.RedoRequest(first.Height)
if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: peerID,
Err: err,
}); serr != nil {
return
}
peerID2 := r.pool.RedoRequest(second.Height)
if peerID2 != peerID {
if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
NodeID: peerID2,
>>>>>>> origin
Err: err,
}); serr != nil {
return
internal/blocksync/reactor_test.go
+2 -4
View File
@@ -190,20 +190,18 @@ func (rts *reactorTestSuite) addNode(
chCreator := func(ctx context.Context, chdesc *p2p.ChannelDescriptor) (*p2p.Channel, error) {
return rts.blockSyncChannels[nodeID], nil
}
rts.reactors[nodeID], err = NewReactor(
ctx,
rts.reactors[nodeID] = NewReactor(
rts.logger.With("nodeID", nodeID),
stateStore,
blockExec,
blockStore,
nil,
chCreator,
rts.peerUpdates[nodeID],
func(ctx context.Context) *p2p.PeerUpdates { return rts.peerUpdates[nodeID] },
rts.blockSync,
consensus.NopMetrics(),
nil, // eventbus, can be nil
)
require.NoError(t, err)
require.NoError(t, rts.reactors[nodeID].Start(ctx))
require.True(t, rts.reactors[nodeID].IsRunning())