Merge remote-tracking branch 'origin' into wb/issue-7950

internal/blocksync/reactor.go

@@ -359,6 +359,13 @@ func (r *Reactor) SwitchToBlockSync(ctx context.Context, state sm.State) error {
 	go r.requestRoutine(ctx, bsCh)
 	go r.poolRoutine(ctx, true, bsCh)
 
+	if err := r.PublishStatus(ctx, types.EventDataBlockSyncStatus{
+		Complete: false,
+		Height:   state.LastBlockHeight,
+	}); err != nil {
+		return err
+	}
+
 	return nil
 }
 

internal/consensus/reactor.go

@@ -752,9 +752,6 @@ func (r *Reactor) gossipVotesForHeight(
 func (r *Reactor) gossipVotesRoutine(ctx context.Context, ps *PeerState, voteCh *p2p.Channel) {
 	logger := r.logger.With("peer", ps.peerID)
 
-	// XXX: simple hack to throttle logs upon sleep
-	logThrottle := 0
-
 	timer := time.NewTimer(0)
 	defer timer.Stop()
 
@@ -772,13 +769,6 @@ func (r *Reactor) gossipVotesRoutine(ctx context.Context, ps *PeerState, voteCh
 		rs := r.getRoundState()
 		prs := ps.GetRoundState()
 
-		switch logThrottle {
-		case 1: // first sleep
-			logThrottle = 2
-		case 2: // no more sleep
-			logThrottle = 0
-		}
-
 		// if height matches, then send LastCommit, Prevotes, and Precommits
 		if rs.Height == prs.Height {
 			if ok, err := r.gossipVotesForHeight(ctx, rs, prs, ps, voteCh); err != nil {
@@ -813,20 +803,6 @@ func (r *Reactor) gossipVotesRoutine(ctx context.Context, ps *PeerState, voteCh
 			}
 		}
 
-		if logThrottle == 0 {
-			// we sent nothing -- sleep
-			logThrottle = 1
-			logger.Debug(
-				"no votes to send; sleeping",
-				"rs.Height", rs.Height,
-				"prs.Height", prs.Height,
-				"localPV", rs.Votes.Prevotes(rs.Round).BitArray(), "peerPV", prs.Prevotes,
-				"localPC", rs.Votes.Precommits(rs.Round).BitArray(), "peerPC", prs.Precommits,
-			)
-		} else if logThrottle == 2 {
-			logThrottle = 1
-		}
-
 		timer.Reset(r.state.config.PeerGossipSleepDuration)
 		select {
 		case <-ctx.Done():
@@ -860,10 +836,20 @@ func (r *Reactor) queryMaj23Routine(ctx context.Context, ps *PeerState, stateCh
 			return
 		}
 
-		rs := r.getRoundState()
-		prs := ps.GetRoundState()
-		// TODO create more reliable coppies of these
+		// TODO create more reliable copies of these
 		// structures so the following go routines don't race
+		rs := r.getRoundState()
+		if rs.Votes == nil {
+			// if we have gotten here, we've connected to
+			// a peer before the state of the reactor has
+			// updated to the current round, so we should
+			// sleep for a while before we attempt to
+			// start gossiping the data that doesn't exist
+			// yet. This prevents a panic.
+			timer.Reset(r.state.config.PeerQueryMaj23SleepDuration)
+			continue
+		}
+		prs := ps.GetRoundState()
 
 		wg := &sync.WaitGroup{}
 

internal/consensus/replay.go

@@ -220,7 +220,6 @@ func NewHandshaker(
 	eventBus *eventbus.EventBus,
 	genDoc *types.GenesisDoc,
 ) *Handshaker {
-
 	return &Handshaker{
 		stateStore:   stateStore,
 		initialState: state,
@@ -228,7 +227,6 @@ func NewHandshaker(
 		eventBus:     eventBus,
 		genDoc:       genDoc,
 		logger:       logger,
-		nBlocks:      0,
 	}
 }
 
@@ -359,7 +357,9 @@ func (h *Handshaker) ReplayBlocks(
 	// First handle edge cases and constraints on the storeBlockHeight and storeBlockBase.
 	switch {
 	case storeBlockHeight == 0:
-		assertAppHashEqualsOneFromState(appHash, state)
+		if err := checkAppHashEqualsOneFromState(appHash, state); err != nil {
+			return nil, err
+		}
 		return appHash, nil
 
 	case appBlockHeight == 0 && state.InitialHeight < storeBlockBase:
@@ -376,11 +376,11 @@ func (h *Handshaker) ReplayBlocks(
 
 	case storeBlockHeight < stateBlockHeight:
 		// the state should never be ahead of the store (this is under tendermint's control)
-		panic(fmt.Sprintf("StateBlockHeight (%d) > StoreBlockHeight (%d)", stateBlockHeight, storeBlockHeight))
+		return nil, fmt.Errorf("StateBlockHeight (%d) > StoreBlockHeight (%d)", stateBlockHeight, storeBlockHeight)
 
 	case storeBlockHeight > stateBlockHeight+1:
 		// store should be at most one ahead of the state (this is under tendermint's control)
-		panic(fmt.Sprintf("StoreBlockHeight (%d) > StateBlockHeight + 1 (%d)", storeBlockHeight, stateBlockHeight+1))
+		return nil, fmt.Errorf("StoreBlockHeight (%d) > StateBlockHeight + 1 (%d)", storeBlockHeight, stateBlockHeight+1)
 	}
 
 	var err error
@@ -395,7 +395,9 @@ func (h *Handshaker) ReplayBlocks(
 
 		} else if appBlockHeight == storeBlockHeight {
 			// We're good!
-			assertAppHashEqualsOneFromState(appHash, state)
+			if err := checkAppHashEqualsOneFromState(appHash, state); err != nil {
+				return nil, err
+			}
 			return appHash, nil
 		}
 
@@ -415,7 +417,11 @@ func (h *Handshaker) ReplayBlocks(
 			// but we'd have to allow the WAL to replay a block that wrote it's #ENDHEIGHT
 			h.logger.Info("Replay last block using real app")
 			state, err = h.replayBlock(ctx, state, storeBlockHeight, appClient)
-			return state.AppHash, err
+			if err != nil {
+				return nil, err
+
+			}
+			return state.AppHash, nil
 
 		case appBlockHeight == storeBlockHeight:
 			// We ran Commit, but didn't save the state, so replayBlock with mock app.
@@ -437,13 +443,13 @@ func (h *Handshaker) ReplayBlocks(
 				return nil, err
 			}
 
-			return state.AppHash, err
+			return state.AppHash, nil
 		}
 
 	}
 
-	panic(fmt.Sprintf("uncovered case! appHeight: %d, storeHeight: %d, stateHeight: %d",
-		appBlockHeight, storeBlockHeight, stateBlockHeight))
+	return nil, fmt.Errorf("uncovered case! appHeight: %d, storeHeight: %d, stateHeight: %d",
+		appBlockHeight, storeBlockHeight, stateBlockHeight)
 }
 
 func (h *Handshaker) replayBlocks(
@@ -452,7 +458,8 @@ func (h *Handshaker) replayBlocks(
 	appClient abciclient.Client,
 	appBlockHeight,
 	storeBlockHeight int64,
-	mutateState bool) ([]byte, error) {
+	mutateState bool,
+) ([]byte, error) {
 	// App is further behind than it should be, so we need to replay blocks.
 	// We replay all blocks from appBlockHeight+1.
 	//
@@ -478,7 +485,9 @@ func (h *Handshaker) replayBlocks(
 		block := h.store.LoadBlock(i)
 		// Extra check to ensure the app was not changed in a way it shouldn't have.
 		if len(appHash) > 0 {
-			assertAppHashEqualsOneFromBlock(appHash, block)
+			if err := checkAppHashEqualsOneFromBlock(appHash, block); err != nil {
+				return nil, err
+			}
 		}
 
 		if i == finalBlock && !mutateState {
@@ -510,7 +519,9 @@ func (h *Handshaker) replayBlocks(
 		appHash = state.AppHash
 	}
 
-	assertAppHashEqualsOneFromState(appHash, state)
+	if err := checkAppHashEqualsOneFromState(appHash, state); err != nil {
+		return nil, err
+	}
 	return appHash, nil
 }
 
@@ -539,24 +550,25 @@ func (h *Handshaker) replayBlock(
 	return state, nil
 }
 
-func assertAppHashEqualsOneFromBlock(appHash []byte, block *types.Block) {
+func checkAppHashEqualsOneFromBlock(appHash []byte, block *types.Block) error {
 	if !bytes.Equal(appHash, block.AppHash) {
-		panic(fmt.Sprintf(`block.AppHash does not match AppHash after replay. Got %X, expected %X.
+		return fmt.Errorf(`block.AppHash does not match AppHash after replay. Got '%X', expected '%X'.
 
-Block: %v
-`,
-			appHash, block.AppHash, block))
+Block: %v`,
+			appHash, block.AppHash, block)
 	}
+	return nil
 }
 
-func assertAppHashEqualsOneFromState(appHash []byte, state sm.State) {
+func checkAppHashEqualsOneFromState(appHash []byte, state sm.State) error {
 	if !bytes.Equal(appHash, state.AppHash) {
-		panic(fmt.Sprintf(`state.AppHash does not match AppHash after replay. Got
-%X, expected %X.
+		return fmt.Errorf(`state.AppHash does not match AppHash after replay. Got '%X', expected '%X'.
 
 State: %v
 
 Did you reset Tendermint without resetting your application's data?`,
-			appHash, state.AppHash, state))
+			appHash, state.AppHash, state)
 	}
+
+	return nil
 }

internal/consensus/replay_test.go

@@ -944,7 +944,7 @@ func buildTMStateFromChain(
 	return state
 }
 
-func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
+func TestHandshakeErrorsIfAppReturnsWrongAppHash(t *testing.T) {
 	// 1. Initialize tendermint and commit 3 blocks with the following app hashes:
 	//		- 0x01
 	//		- 0x02
@@ -988,12 +988,8 @@ func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
 		require.NoError(t, err)
 		t.Cleanup(func() { cancel(); proxyApp.Wait() })
 
-		assert.Panics(t, func() {
-			h := NewHandshaker(logger, stateStore, state, store, eventBus, genDoc)
-			if err = h.Handshake(ctx, proxyApp); err != nil {
-				t.Log(err)
-			}
-		})
+		h := NewHandshaker(logger, stateStore, state, store, eventBus, genDoc)
+		assert.Error(t, h.Handshake(ctx, proxyApp))
 	}
 
 	// 3. Tendermint must panic if app returns wrong hash for the last block
@@ -1008,12 +1004,8 @@ func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
 		require.NoError(t, err)
 		t.Cleanup(func() { cancel(); proxyApp.Wait() })
 
-		assert.Panics(t, func() {
-			h := NewHandshaker(logger, stateStore, state, store, eventBus, genDoc)
-			if err = h.Handshake(ctx, proxyApp); err != nil {
-				t.Log(err)
-			}
-		})
+		h := NewHandshaker(logger, stateStore, state, store, eventBus, genDoc)
+		require.Error(t, h.Handshake(ctx, proxyApp))
 	}
 }
 

internal/consensus/state.go

@@ -124,6 +124,7 @@ type State struct {
 
 	stateStore            sm.Store
 	initialStatePopulated bool
+	skipBootstrapping     bool
 
 	// create and execute blocks
 	blockExec *sm.BlockExecutor
@@ -185,6 +186,12 @@ type State struct {
 // StateOption sets an optional parameter on the State.
 type StateOption func(*State)
 
+// SkipStateStoreBootstrap is a state option forces the constructor to
+// skip state bootstrapping during construction.
+func SkipStateStoreBootstrap(sm *State) {
+	sm.skipBootstrapping = true
+}
+
 // NewState returns a new State.
 func NewState(
 	ctx context.Context,
@@ -223,16 +230,21 @@ func NewState(
 	cs.doPrevote = cs.defaultDoPrevote
 	cs.setProposal = cs.defaultSetProposal
 
-	if err := cs.updateStateFromStore(ctx); err != nil {
-		return nil, err
-	}
-
 	// NOTE: we do not call scheduleRound0 yet, we do that upon Start()
 	cs.BaseService = *service.NewBaseService(logger, "State", cs)
 	for _, option := range options {
 		option(cs)
 	}
 
+	// this is not ideal, but it lets the consensus tests start
+	// node-fragments gracefully while letting the nodes
+	// themselves avoid this.
+	if !cs.skipBootstrapping {
+		if err := cs.updateStateFromStore(ctx); err != nil {
+			return nil, err
+		}
+	}
+
 	return cs, nil
 }
 

internal/p2p/README.md

@@ -7,5 +7,5 @@ Docs:
 - [Connection](https://docs.tendermint.com/master/spec/p2p/connection.html) for details on how connections and multiplexing work
 - [Peer](https://docs.tendermint.com/master/spec/p2p/node.html) for details on peer ID, handshakes, and peer exchange
 - [Node](https://docs.tendermint.com/master/spec/p2p/node.html) for details about different types of nodes and how they should work
-- [Pex](https://docs.tendermint.com/master/spec/reactors/pex/pex.html) for details on peer discovery and exchange
+- [Pex](https://docs.tendermint.com/master/spec/p2p/messages/pex.html) for details on peer discovery and exchange
 - [Config](https://docs.tendermint.com/master/spec/p2p/config.html) for details on some config option

internal/rpc/core/status.go

@@ -77,14 +77,25 @@ func (env *Environment) Status(ctx context.Context) (*coretypes.ResultStatus, er
 			EarliestAppHash:     earliestAppHash,
 			EarliestBlockHeight: earliestBlockHeight,
 			EarliestBlockTime:   time.Unix(0, earliestBlockTimeNano),
-			MaxPeerBlockHeight:  env.BlockSyncReactor.GetMaxPeerBlockHeight(),
-			CatchingUp:          env.ConsensusReactor.WaitSync(),
-			TotalSyncedTime:     env.BlockSyncReactor.GetTotalSyncedTime(),
-			RemainingTime:       env.BlockSyncReactor.GetRemainingSyncTime(),
+			// this should start as true, if consensus
+			// hasn't started yet, and then flip to false
+			// (or true,) depending on what's actually
+			// happening.
+			CatchingUp: true,
 		},
 		ValidatorInfo: validatorInfo,
 	}
 
+	if env.ConsensusReactor != nil {
+		result.SyncInfo.CatchingUp = env.ConsensusReactor.WaitSync()
+	}
+
+	if env.BlockSyncReactor != nil {
+		result.SyncInfo.MaxPeerBlockHeight = env.BlockSyncReactor.GetMaxPeerBlockHeight()
+		result.SyncInfo.TotalSyncedTime = env.BlockSyncReactor.GetTotalSyncedTime()
+		result.SyncInfo.RemainingTime = env.BlockSyncReactor.GetRemainingSyncTime()
+	}
+
 	if env.StateSyncMetricer != nil {
 		result.SyncInfo.TotalSnapshots = env.StateSyncMetricer.TotalSnapshots()
 		result.SyncInfo.ChunkProcessAvgTime = env.StateSyncMetricer.ChunkProcessAvgTime()
@@ -103,6 +114,9 @@ func (env *Environment) validatorAtHeight(h int64) *types.Validator {
 	if err != nil {
 		return nil
 	}
+	if env.ConsensusState == nil {
+		return nil
+	}
 	if env.PubKey == nil {
 		return nil
 	}

internal/statesync/reactor.go

@@ -143,6 +143,12 @@ type Reactor struct {
 	peerEvents     p2p.PeerEventSubscriber
 	chCreator      p2p.ChannelCreator
 	sendBlockError func(context.Context, p2p.PeerError) error
+	postSyncHook   func(context.Context, sm.State) error
+
+	// when true, the reactor will, during startup perform a
+	// statesync for this node, and otherwise just provide
+	// snapshots to other nodes.
+	needsStateSync bool
 
 	// Dispatcher is used to multiplex light block requests and responses over multiple
 	// peers used by the p2p state provider and in reverse sync.
@@ -171,7 +177,6 @@ type Reactor struct {
 // and querying, references to p2p Channels and a channel to listen for peer
 // updates on. Note, the reactor will close all p2p Channels when stopping.
 func NewReactor(
-	ctx context.Context,
 	chainID string,
 	initialHeight int64,
 	cfg config.StateSyncConfig,
@@ -184,23 +189,26 @@ func NewReactor(
 	tempDir string,
 	ssMetrics *Metrics,
 	eventBus *eventbus.EventBus,
+	postSyncHook func(context.Context, sm.State) error,
+	needsStateSync bool,
 ) *Reactor {
-
 	r := &Reactor{
-		logger:        logger,
-		chainID:       chainID,
-		initialHeight: initialHeight,
-		cfg:           cfg,
-		conn:          conn,
-		chCreator:     channelCreator,
-		peerEvents:    peerEvents,
-		tempDir:       tempDir,
-		stateStore:    stateStore,
-		blockStore:    blockStore,
-		peers:         newPeerList(),
-		providers:     make(map[types.NodeID]*BlockProvider),
-		metrics:       ssMetrics,
-		eventBus:      eventBus,
+		logger:         logger,
+		chainID:        chainID,
+		initialHeight:  initialHeight,
+		cfg:            cfg,
+		conn:           conn,
+		chCreator:      channelCreator,
+		peerEvents:     peerEvents,
+		tempDir:        tempDir,
+		stateStore:     stateStore,
+		blockStore:     blockStore,
+		peers:          newPeerList(),
+		providers:      make(map[types.NodeID]*BlockProvider),
+		metrics:        ssMetrics,
+		eventBus:       eventBus,
+		postSyncHook:   postSyncHook,
+		needsStateSync: needsStateSync,
 	}
 
 	r.BaseService = *service.NewBaseService(logger, "StateSync", r)
@@ -300,6 +308,14 @@ func (r *Reactor) OnStart(ctx context.Context) error {
 	go r.processChannels(ctx, snapshotCh, chunkCh, blockCh, paramsCh)
 	go r.processPeerUpdates(ctx, r.peerEvents(ctx))
 
+	if r.needsStateSync {
+		r.logger.Info("starting state sync")
+		if _, err := r.Sync(ctx); err != nil {
+			r.logger.Error("state sync failed; shutting down this node", "err", err)
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -310,20 +326,21 @@ func (r *Reactor) OnStop() {
 	r.dispatcher.Close()
 }
 
-func (r *Reactor) PublishStatus(ctx context.Context, event types.EventDataStateSyncStatus) error {
-	if r.eventBus == nil {
-		return errors.New("event system is not configured")
-	}
-
-	return r.eventBus.PublishEventStateSyncStatus(ctx, event)
-}
-
 // Sync runs a state sync, fetching snapshots and providing chunks to the
 // application. At the close of the operation, Sync will bootstrap the state
 // store and persist the commit at that height so that either consensus or
 // blocksync can commence. It will then proceed to backfill the necessary amount
 // of historical blocks before participating in consensus
 func (r *Reactor) Sync(ctx context.Context) (sm.State, error) {
+	if r.eventBus != nil {
+		if err := r.eventBus.PublishEventStateSyncStatus(ctx, types.EventDataStateSyncStatus{
+			Complete: false,
+			Height:   r.initialHeight,
+		}); err != nil {
+			return sm.State{}, err
+		}
+	}
+
 	// We need at least two peers (for cross-referencing of light blocks) before we can
 	// begin state sync
 	if err := r.waitForEnoughPeers(ctx, 2); err != nil {
@@ -357,21 +374,33 @@ func (r *Reactor) Sync(ctx context.Context) (sm.State, error) {
 		return sm.State{}, err
 	}
 
-	err = r.stateStore.Bootstrap(state)
-	if err != nil {
+	if err := r.stateStore.Bootstrap(state); err != nil {
 		return sm.State{}, fmt.Errorf("failed to bootstrap node with new state: %w", err)
 	}
 
-	err = r.blockStore.SaveSeenCommit(state.LastBlockHeight, commit)
-	if err != nil {
+	if err := r.blockStore.SaveSeenCommit(state.LastBlockHeight, commit); err != nil {
 		return sm.State{}, fmt.Errorf("failed to store last seen commit: %w", err)
 	}
 
-	err = r.Backfill(ctx, state)
-	if err != nil {
+	if err := r.Backfill(ctx, state); err != nil {
 		r.logger.Error("backfill failed. Proceeding optimistically...", "err", err)
 	}
 
+	if r.eventBus != nil {
+		if err := r.eventBus.PublishEventStateSyncStatus(ctx, types.EventDataStateSyncStatus{
+			Complete: true,
+			Height:   state.LastBlockHeight,
+		}); err != nil {
+			return sm.State{}, err
+		}
+	}
+
+	if r.postSyncHook != nil {
+		if err := r.postSyncHook(ctx, state); err != nil {
+			return sm.State{}, err
+		}
+	}
+
 	return state, nil
 }
 

internal/statesync/reactor_test.go

@@ -155,7 +155,6 @@ func setup(
 	logger := log.NewNopLogger()
 
 	rts.reactor = NewReactor(
-		ctx,
 		factory.DefaultTestChainID,
 		1,
 		*cfg,
@@ -167,7 +166,9 @@ func setup(
 		rts.blockStore,
 		"",
 		m,
-		nil, // eventbus can be nil
+		nil,   // eventbus can be nil
+		nil,   // post-sync-hook
+		false, // run Sync during Start()
 	)
 
 	rts.syncer = &syncer{