Merge remote-tracking branch 'origin/master' into wb/abci-finalize-block-synchronize

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

@@ -370,7 +370,11 @@ func subscribeToVoter(ctx context.Context, t *testing.T, cs *State, addr []byte)
 		vote := msg.Data().(types.EventDataVote)
 		// we only fire for our own votes
 		if bytes.Equal(addr, vote.Vote.ValidatorAddress) {
-			ch <- msg
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case ch <- msg:
+			}
 		}
 		return nil
 	}, types.EventQueryVote); err != nil {
@@ -401,7 +405,10 @@ func subscribeToVoterBuffered(ctx context.Context, t *testing.T, cs *State, addr
 			vote := msg.Data().(types.EventDataVote)
 			// we only fire for our own votes
 			if bytes.Equal(addr, vote.Vote.ValidatorAddress) {
-				ch <- msg
+				select {
+				case <-ctx.Done():
+				case ch <- msg:
+				}
 			}
 		}
 	}()
@@ -462,7 +469,6 @@ func newStateWithConfigAndBlockStore(
 		logger.With("module", "mempool"),
 		thisConfig.Mempool,
 		proxyAppConnMem,
-		0,
 	)
 
 	if thisConfig.Consensus.WaitForTxs() {
@@ -477,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.Consensus(), 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

@@ -30,8 +30,10 @@ import (
 // stripped down version of node (proxy app, event bus, consensus state) with a
 // persistent kvstore application and special consensus wal instance
 // (byteBufferWAL) and waits until numBlocks are created.
-// If the node fails to produce given numBlocks, it returns an error.
-func WALGenerateNBlocks(ctx context.Context, t *testing.T, logger log.Logger, wr io.Writer, numBlocks int) (err error) {
+// If the node fails to produce given numBlocks, it fails the test.
+func WALGenerateNBlocks(ctx context.Context, t *testing.T, logger log.Logger, wr io.Writer, numBlocks int) {
+	t.Helper()
+
 	cfg := getConfig(t)
 
 	app := kvstore.NewPersistentKVStoreApplication(logger, filepath.Join(cfg.DBDir(), "wal_generator"))
@@ -46,40 +48,46 @@ func WALGenerateNBlocks(ctx context.Context, t *testing.T, logger log.Logger, wr
 	privValidatorStateFile := cfg.PrivValidator.StateFile()
 	privValidator, err := privval.LoadOrGenFilePV(privValidatorKeyFile, privValidatorStateFile)
 	if err != nil {
-		return err
+		t.Fatal(err)
 	}
 	genDoc, err := types.GenesisDocFromFile(cfg.GenesisFile())
 	if err != nil {
-		return fmt.Errorf("failed to read genesis file: %w", err)
+		t.Fatal(fmt.Errorf("failed to read genesis file: %w", err))
 	}
 	blockStoreDB := dbm.NewMemDB()
 	stateDB := blockStoreDB
 	stateStore := sm.NewStore(stateDB)
 	state, err := sm.MakeGenesisState(genDoc)
 	if err != nil {
-		return fmt.Errorf("failed to make genesis state: %w", err)
+		t.Fatal(fmt.Errorf("failed to make genesis state: %w", err))
 	}
 	state.Version.Consensus.App = kvstore.ProtocolVersion
 	if err = stateStore.Save(state); err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 
 	blockStore := store.NewBlockStore(blockStoreDB)
 
 	proxyApp := proxy.NewAppConns(abciclient.NewLocalCreator(app), logger.With("module", "proxy"), proxy.NopMetrics())
 	if err := proxyApp.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start proxy app connections: %w", err)
+		t.Fatal(fmt.Errorf("failed to start proxy app connections: %w", err))
 	}
+	t.Cleanup(proxyApp.Wait)
 
 	eventBus := eventbus.NewDefault(logger.With("module", "events"))
 	if err := eventBus.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start event bus: %w", err)
+		t.Fatal(fmt.Errorf("failed to start event bus: %w", err))
 	}
+	t.Cleanup(func() { eventBus.Stop(); eventBus.Wait() })
 
 	mempool := emptyMempool{}
 	evpool := sm.EmptyEvidencePool{}
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp.Consensus(), 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)
@@ -91,22 +99,24 @@ func WALGenerateNBlocks(ctx context.Context, t *testing.T, logger log.Logger, wr
 	wal := newByteBufferWAL(logger, NewWALEncoder(wr), int64(numBlocks), numBlocksWritten)
 	// see wal.go#103
 	if err := wal.Write(EndHeightMessage{0}); err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 
 	consensusState.wal = wal
 
 	if err := consensusState.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start consensus state: %w", err)
+		t.Fatal(fmt.Errorf("failed to start consensus state: %w", err))
 	}
+	t.Cleanup(consensusState.Wait)
+
+	defer consensusState.Stop()
+	timer := time.NewTimer(time.Minute)
+	defer timer.Stop()
 
 	select {
 	case <-numBlocksWritten:
-		consensusState.Stop()
-		return nil
-	case <-time.After(1 * time.Minute):
-		consensusState.Stop()
-		return fmt.Errorf("waited too long for tendermint to produce %d blocks (grep logs for `wal_generator`)", numBlocks)
+	case <-timer.C:
+		t.Fatal(fmt.Errorf("waited too long for tendermint to produce %d blocks (grep logs for `wal_generator`)", numBlocks))
 	}
 }
 
@@ -115,9 +125,7 @@ func WALWithNBlocks(ctx context.Context, t *testing.T, logger log.Logger, numBlo
 	var b bytes.Buffer
 	wr := bufio.NewWriter(&b)
 
-	if err := WALGenerateNBlocks(ctx, t, logger, wr, numBlocks); err != nil {
-		return []byte{}, err
-	}
+	WALGenerateNBlocks(ctx, t, logger, wr, numBlocks)
 
 	wr.Flush()
 	return b.Bytes(), nil

internal/consensus/wal_test.go

@@ -3,6 +3,7 @@ package consensus
 import (
 	"bytes"
 	"context"
+	"os"
 	"path/filepath"
 
 	"testing"
@@ -41,13 +42,12 @@ func TestWALTruncate(t *testing.T) {
 	require.NoError(t, err)
 	err = wal.Start(ctx)
 	require.NoError(t, err)
-	t.Cleanup(wal.Wait)
+	t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() })
 
 	// 60 block's size nearly 70K, greater than group's headBuf size(4096 * 10),
 	// when headBuf is full, truncate content will Flush to the file. at this
 	// time, RotateFile is called, truncate content exist in each file.
-	err = WALGenerateNBlocks(ctx, t, logger, wal.Group(), 60)
-	require.NoError(t, err)
+	WALGenerateNBlocks(ctx, t, logger, wal.Group(), 60)
 
 	// put the leakcheck here so it runs after other cleanup
 	// functions.
@@ -112,7 +112,7 @@ func TestWALWrite(t *testing.T) {
 	require.NoError(t, err)
 	err = wal.Start(ctx)
 	require.NoError(t, err)
-	t.Cleanup(wal.Wait)
+	t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() })
 
 	// 1) Write returns an error if msg is too big
 	msg := &BlockPartMessage{
@@ -151,7 +151,6 @@ func TestWALSearchForEndHeight(t *testing.T) {
 
 	wal, err := NewWAL(ctx, logger, walFile)
 	require.NoError(t, err)
-	t.Cleanup(func() { wal.Stop(); wal.Wait() })
 
 	h := int64(3)
 	gr, found, err := wal.SearchForEndHeight(h, &WALSearchOptions{})
@@ -176,24 +175,24 @@ func TestWALPeriodicSync(t *testing.T) {
 
 	walDir := t.TempDir()
 	walFile := filepath.Join(walDir, "wal")
-	wal, err := NewWAL(ctx, log.TestingLogger(), walFile, autofile.GroupCheckDuration(1*time.Millisecond))
+	defer os.RemoveAll(walFile)
 
+	wal, err := NewWAL(ctx, log.TestingLogger(), walFile, autofile.GroupCheckDuration(250*time.Millisecond))
 	require.NoError(t, err)
 
 	wal.SetFlushInterval(walTestFlushInterval)
 	logger := log.NewNopLogger()
 
 	// Generate some data
-	err = WALGenerateNBlocks(ctx, t, logger, wal.Group(), 5)
-	require.NoError(t, err)
+	WALGenerateNBlocks(ctx, t, logger, wal.Group(), 5)
 
 	// We should have data in the buffer now
 	assert.NotZero(t, wal.Group().Buffered())
 
 	require.NoError(t, wal.Start(ctx))
-	t.Cleanup(func() { wal.Stop(); wal.Wait() })
+	t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() })
 
-	time.Sleep(walTestFlushInterval + (10 * time.Millisecond))
+	time.Sleep(walTestFlushInterval + (20 * time.Millisecond))
 
 	// The data should have been flushed by the periodic sync
 	assert.Zero(t, wal.Group().Buffered())