Merge remote-tracking branch 'origin/master' into proxy-remove-triforcated-client

internal/consensus/byzantine_test.go

@@ -82,7 +82,6 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 				log.TestingLogger().With("module", "mempool"),
 				thisConfig.Mempool,
 				proxyAppConnMem,
-				0,
 			)
 			if thisConfig.Consensus.WaitForTxs() {
 				mempool.EnableTxsAvailable()
@@ -95,7 +94,8 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 			// Make State
 			blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool, blockStore)
-			cs := NewState(ctx, logger, thisConfig.Consensus, state, blockExec, blockStore, mempool, evpool)
+			cs, err := NewState(ctx, logger, thisConfig.Consensus, stateStore, blockExec, blockStore, mempool, evpool)
+			require.NoError(t, err)
 			// set private validator
 			pv := privVals[i]
 			cs.SetPrivValidator(ctx, pv)
@@ -105,14 +105,13 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 			require.NoError(t, err)
 			cs.SetEventBus(eventBus)
 			evpool.SetEventBus(eventBus)
-
 			cs.SetTimeoutTicker(tickerFunc())
 
 			states[i] = cs
 		}()
 	}
 
-	rts := setup(ctx, t, nValidators, states, 100) // buffer must be large enough to not deadlock
+	rts := setup(ctx, t, nValidators, states, 512) // buffer must be large enough to not deadlock
 
 	var bzNodeID types.NodeID
 
@@ -238,8 +237,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 	}
 
 	for _, reactor := range rts.reactors {
-		state := reactor.state.GetState()
-		reactor.SwitchToConsensus(ctx, state, false)
+		reactor.SwitchToConsensus(ctx, reactor.state.GetState(), false)
 	}
 
 	// Evidence should be submitted and committed at the third height but
@@ -248,20 +246,26 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 	var wg sync.WaitGroup
 	i := 0
+	subctx, subcancel := context.WithCancel(ctx)
+	defer subcancel()
 	for _, sub := range rts.subs {
 		wg.Add(1)
 
 		go func(j int, s eventbus.Subscription) {
 			defer wg.Done()
 			for {
-				if ctx.Err() != nil {
+				if subctx.Err() != nil {
+					return
+				}
+
+				msg, err := s.Next(subctx)
+				if subctx.Err() != nil {
 					return
 				}
 
-				msg, err := s.Next(ctx)
-				assert.NoError(t, err)
 				if err != nil {
-					cancel()
+					t.Errorf("waiting for subscription: %v", err)
+					subcancel()
 					return
 				}
 
@@ -273,12 +277,18 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 				}
 			}
 		}(i, sub)
-
 		i++
 	}
 
 	wg.Wait()
 
+	// don't run more assertions if we've encountered a timeout
+	select {
+	case <-subctx.Done():
+		t.Fatal("encountered timeout")
+	default:
+	}
+
 	pubkey, err := bzNodeState.privValidator.GetPubKey(ctx)
 	require.NoError(t, err)
 

internal/consensus/common_test.go

@@ -469,7 +469,6 @@ func newStateWithConfigAndBlockStore(
 		logger.With("module", "mempool"),
 		thisConfig.Mempool,
 		proxyAppConnMem,
-		0,
 	)
 
 	if thisConfig.Consensus.WaitForTxs() {
@@ -484,15 +483,19 @@ func newStateWithConfigAndBlockStore(
 	require.NoError(t, stateStore.Save(state))
 
 	blockExec := sm.NewBlockExecutor(stateStore, logger, proxyAppConnCon, mempool, evpool, blockStore)
-	cs := NewState(ctx,
+	cs, err := NewState(ctx,
 		logger.With("module", "consensus"),
 		thisConfig.Consensus,
-		state,
+		stateStore,
 		blockExec,
 		blockStore,
 		mempool,
 		evpool,
 	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	cs.SetPrivValidator(ctx, pv)
 
 	eventBus := eventbus.NewDefault(logger.With("module", "events"))

internal/consensus/reactor_test.go

@@ -461,6 +461,7 @@ func TestReactorWithEvidence(t *testing.T) {
 		stateStore := sm.NewStore(stateDB)
 		state, err := sm.MakeGenesisState(genDoc)
 		require.NoError(t, err)
+		require.NoError(t, stateStore.Save(state))
 		thisConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%d", testName, i))
 		require.NoError(t, err)
 
@@ -483,7 +484,6 @@ func TestReactorWithEvidence(t *testing.T) {
 			log.TestingLogger().With("module", "mempool"),
 			thisConfig.Mempool,
 			proxyAppConnMem,
-			0,
 		)
 
 		if thisConfig.Consensus.WaitForTxs() {
@@ -506,8 +506,9 @@ func TestReactorWithEvidence(t *testing.T) {
 
 		blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool, blockStore)
 
-		cs := NewState(ctx, logger.With("validator", i, "module", "consensus"),
-			thisConfig.Consensus, state, blockExec, blockStore, mempool, evpool2)
+		cs, err := NewState(ctx, logger.With("validator", i, "module", "consensus"),
+			thisConfig.Consensus, stateStore, blockExec, blockStore, mempool, evpool2)
+		require.NoError(t, err)
 		cs.SetPrivValidator(ctx, pv)
 
 		eventBus := eventbus.NewDefault(log.TestingLogger().With("module", "events"))

internal/consensus/replay_file.go

@@ -84,7 +84,7 @@ func (cs *State) ReplayFile(ctx context.Context, file string, console bool) erro
 		return err
 	}
 
-	pb := newPlayback(file, fp, cs, cs.state.Copy())
+	pb := newPlayback(file, fp, cs, cs.stateStore)
 	defer pb.fp.Close()
 
 	var nextN int // apply N msgs in a row
@@ -126,17 +126,17 @@ type playback struct {
 	count int // how many lines/msgs into the file are we
 
 	// replays can be reset to beginning
-	fileName     string   // so we can close/reopen the file
-	genesisState sm.State // so the replay session knows where to restart from
+	fileName   string // so we can close/reopen the file
+	stateStore sm.Store
 }
 
-func newPlayback(fileName string, fp *os.File, cs *State, genState sm.State) *playback {
+func newPlayback(fileName string, fp *os.File, cs *State, store sm.Store) *playback {
 	return &playback{
-		cs:           cs,
-		fp:           fp,
-		fileName:     fileName,
-		genesisState: genState,
-		dec:          NewWALDecoder(fp),
+		cs:         cs,
+		fp:         fp,
+		fileName:   fileName,
+		stateStore: store,
+		dec:        NewWALDecoder(fp),
 	}
 }
 
@@ -145,8 +145,11 @@ func (pb *playback) replayReset(ctx context.Context, count int, newStepSub event
 	pb.cs.Stop()
 	pb.cs.Wait()
 
-	newCS := NewState(ctx, pb.cs.logger, pb.cs.config, pb.genesisState.Copy(), pb.cs.blockExec,
+	newCS, err := NewState(ctx, pb.cs.logger, pb.cs.config, pb.stateStore, pb.cs.blockExec,
 		pb.cs.blockStore, pb.cs.txNotifier, pb.cs.evpool)
+	if err != nil {
+		return err
+	}
 	newCS.SetEventBus(pb.cs.eventBus)
 	newCS.startForReplay()
 
@@ -345,9 +348,11 @@ func newConsensusStateForReplay(
 	mempool, evpool := emptyMempool{}, sm.EmptyEvidencePool{}
 	blockExec := sm.NewBlockExecutor(stateStore, logger, proxyApp, mempool, evpool, blockStore)
 
-	consensusState := NewState(ctx, logger, csConfig, state.Copy(), blockExec,
+	consensusState, err := NewState(ctx, logger, csConfig, stateStore, blockExec,
 		blockStore, mempool, evpool)
-
+	if err != nil {
+		return nil, err
+	}
 	consensusState.SetEventBus(eventBus)
 	return consensusState, nil
 }

internal/consensus/state.go

@@ -121,6 +121,9 @@ type State struct {
 	// store blocks and commits
 	blockStore sm.BlockStore
 
+	stateStore            sm.Store
+	initialStatePopulated bool
+
 	// create and execute blocks
 	blockExec *sm.BlockExecutor
 
@@ -189,18 +192,19 @@ func NewState(
 	ctx context.Context,
 	logger log.Logger,
 	cfg *config.ConsensusConfig,
-	state sm.State,
+	store sm.Store,
 	blockExec *sm.BlockExecutor,
 	blockStore sm.BlockStore,
 	txNotifier txNotifier,
 	evpool evidencePool,
 	options ...StateOption,
-) *State {
+) (*State, error) {
 	cs := &State{
 		logger:           logger,
 		config:           cfg,
 		blockExec:        blockExec,
 		blockStore:       blockStore,
+		stateStore:       store,
 		txNotifier:       txNotifier,
 		peerMsgQueue:     make(chan msgInfo, msgQueueSize),
 		internalMsgQueue: make(chan msgInfo, msgQueueSize),
@@ -220,6 +224,31 @@ 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)
+	}
+
+	return cs, nil
+}
+
+func (cs *State) updateStateFromStore(ctx context.Context) error {
+	if cs.initialStatePopulated {
+		return nil
+	}
+	state, err := cs.stateStore.Load()
+	if err != nil {
+		return fmt.Errorf("loading state: %w", err)
+	}
+	if state.IsEmpty() {
+		return nil
+	}
+
 	// We have no votes, so reconstruct LastCommit from SeenCommit.
 	if state.LastBlockHeight > 0 {
 		cs.reconstructLastCommit(state)
@@ -227,14 +256,8 @@ func NewState(
 
 	cs.updateToState(ctx, state)
 
-	// 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)
-	}
-
-	return cs
+	cs.initialStatePopulated = true
+	return nil
 }
 
 // SetEventBus sets event bus.
@@ -365,6 +388,10 @@ func (cs *State) LoadCommit(height int64) *types.Commit {
 // OnStart loads the latest state via the WAL, and starts the timeout and
 // receive routines.
 func (cs *State) OnStart(ctx context.Context) error {
+	if err := cs.updateStateFromStore(ctx); err != nil {
+		return err
+	}
+
 	// We may set the WAL in testing before calling Start, so only OpenWAL if its
 	// still the nilWAL.
 	if _, ok := cs.wal.(nilWAL); ok {

internal/consensus/wal_generator.go

@@ -83,7 +83,11 @@ func WALGenerateNBlocks(ctx context.Context, t *testing.T, logger log.Logger, wr
 	mempool := emptyMempool{}
 	evpool := sm.EmptyEvidencePool{}
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp, mempool, evpool, blockStore)
-	consensusState := NewState(ctx, logger, cfg.Consensus, state.Copy(), blockExec, blockStore, mempool, evpool)
+	consensusState, err := NewState(ctx, logger, cfg.Consensus, stateStore, blockExec, blockStore, mempool, evpool)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	consensusState.SetEventBus(eventBus)
 	if privValidator != nil && privValidator != (*privval.FilePV)(nil) {
 		consensusState.SetPrivValidator(ctx, privValidator)