func makeBlockchainFromWAL(wal WAL) ([]*types.Block, []*types.Commit, error) {
	gr, found, err := wal.SearchForEndHeight(0, &WALSearchOptions{})
	if err != nil {
		return nil, nil, err
	}
	if !found {
		return nil, nil, fmt.Errorf("WAL does not contain height %d.", 1)
	}
	defer gr.Close()  
	var blocks []*types.Block
	var commits []*types.Commit
	var thisBlockParts *types.PartSet
	var thisBlockCommit *types.Commit
	var height int64
	dec := NewWALDecoder(gr)
	for {
		msg, err := dec.Decode()
		if err == io.EOF {
			break
		} else if err != nil {
			return nil, nil, err
		}
		piece := readPieceFromWAL(msg)
		if piece == nil {
			continue
		}
		switch p := piece.(type) {
		case EndHeightMessage:
			if thisBlockParts != nil {
				var block = new(types.Block)
				_, err = cdc.UnmarshalBinaryLengthPrefixedReader(thisBlockParts.GetReader(), block, 0)
				if err != nil {
					panic(err)
				}
				if block.Height != height+1 {
					panic(fmt.Sprintf("read bad block from wal. got height %d, expected %d", block.Height, height+1))
				}
				commitHeight := thisBlockCommit.Precommits[0].Height
				if commitHeight != height+1 {
					panic(fmt.Sprintf("commit doesnt match. got height %d, expected %d", commitHeight, height+1))
				}
				blocks = append(blocks, block)
				commits = append(commits, thisBlockCommit)
				height++
			}
		case *types.PartSetHeader:
			thisBlockParts = types.NewPartSetFromHeader(*p)
		case *types.Part:
			_, err := thisBlockParts.AddPart(p)
			if err != nil {
				return nil, nil, err
			}
		case *types.Vote:
			if p.Type == types.PrecommitType {
				commitSigs := []*types.CommitSig{p.CommitSig()}
				thisBlockCommit = types.NewCommit(p.BlockID, commitSigs)
			}
		}
	}
	var block = new(types.Block)
	_, err = cdc.UnmarshalBinaryLengthPrefixedReader(thisBlockParts.GetReader(), block, 0)
	if err != nil {
		panic(err)
	}
	if block.Height != height+1 {
		panic(fmt.Sprintf("read bad block from wal. got height %d, expected %d", block.Height, height+1))
	}
	commitHeight := thisBlockCommit.Precommits[0].Height
	if commitHeight != height+1 {
		panic(fmt.Sprintf("commit doesnt match. got height %d, expected %d", commitHeight, height+1))
	}
	blocks = append(blocks, block)
	commits = append(commits, thisBlockCommit)
	return blocks, commits, nil
}
--------------------------
utils for making blocks
package consensus
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path"
	"runtime"
	"testing"
	"time"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/XuanMaoSecLab/DolphinChain/abci/example/kvstore"
	abci "github.com/XuanMaoSecLab/DolphinChain/abci/types"
	cfg "github.com/XuanMaoSecLab/DolphinChain/config"
	"github.com/XuanMaoSecLab/DolphinChain/crypto"
	dbm "github.com/XuanMaoSecLab/DolphinChain/libs/db"
	"github.com/XuanMaoSecLab/DolphinChain/libs/log"
	"github.com/XuanMaoSecLab/DolphinChain/privval"
	"github.com/XuanMaoSecLab/DolphinChain/proxy"
	sm "github.com/XuanMaoSecLab/DolphinChain/state"
	"github.com/XuanMaoSecLab/DolphinChain/types"
	"github.com/XuanMaoSecLab/DolphinChain/version"
)
func TestMain(m *testing.M) {
	config = ResetConfig("consensus_reactor_test")
	consensusReplayConfig = ResetConfig("consensus_replay_test")
	configStateTest := ResetConfig("consensus_state_test")
	configMempoolTest := ResetConfig("consensus_mempool_test")
	configByzantineTest := ResetConfig("consensus_byzantine_test")
	code := m.Run()
	os.RemoveAll(config.RootDir)
	os.RemoveAll(consensusReplayConfig.RootDir)
	os.RemoveAll(configStateTest.RootDir)
	os.RemoveAll(configMempoolTest.RootDir)
	os.RemoveAll(configByzantineTest.RootDir)
	os.Exit(code)
}
func startNewConsensusStateAndWaitForBlock(t *testing.T, consensusReplayConfig *cfg.Config,
	lastBlockHeight int64, blockDB dbm.DB, stateDB dbm.DB) {
	logger := log.TestingLogger()
	state, _ := sm.LoadStateFromDBOrGenesisFile(stateDB, consensusReplayConfig.GenesisFile())
	privValidator := loadPrivValidator(consensusReplayConfig)
	cs := newConsensusStateWithConfigAndBlockStore(consensusReplayConfig, state, privValidator, kvstore.NewKVStoreApplication(), blockDB)
	cs.SetLogger(logger)
	bytes, _ := ioutil.ReadFile(cs.config.WalFile())
	t.Logf("====== WAL: \n\r%X\n", bytes)
	err := cs.Start()
	require.NoError(t, err)
	defer cs.Stop()
	newBlockSub, err := cs.eventBus.Subscribe(context.Background(), testSubscriber, types.EventQueryNewBlock)
	require.NoError(t, err)
	select {
	case <-newBlockSub.Out():
	case <-newBlockSub.Cancelled():
		t.Fatal("newBlockSub was cancelled")
	case <-time.After(120 * time.Second):
		t.Fatal("Timed out waiting for new block (see trace above)")
	}
}
func sendTxs(cs *ConsensusState, ctx context.Context) {
	for i := 0; i < 256; i++ {
		select {
		case <-ctx.Done():
			return
		default:
			tx := []byte{byte(i)}
			assertMempool(cs.txNotifier).CheckTx(tx, nil)
			i++
		}
	}
}
func TestWALCrash(t *testing.T) {
	testCases := []struct {
		name         string
		initFn       func(dbm.DB, *ConsensusState, context.Context)
		heightToStop int64
	}{
		{"empty block",
			func(stateDB dbm.DB, cs *ConsensusState, ctx context.Context) {},
			1},
		{"many non-empty blocks",
			func(stateDB dbm.DB, cs *ConsensusState, ctx context.Context) {
				go sendTxs(cs, ctx)
			},
			3},
	}
	for i, tc := range testCases {
		consensusReplayConfig := ResetConfig(fmt.Sprintf("%s_%d", t.Name(), i))
		t.Run(tc.name, func(t *testing.T) {
			crashWALandCheckLiveness(t, consensusReplayConfig, tc.initFn, tc.heightToStop)
		})
	}
}
func crashWALandCheckLiveness(t *testing.T, consensusReplayConfig *cfg.Config,
	initFn func(dbm.DB, *ConsensusState, context.Context), heightToStop int64) {
	walPanicked := make(chan error)
	crashingWal := &crashingWAL{panicCh: walPanicked, heightToStop: heightToStop}
	i := 1
LOOP:
	for {
		t.Logf("====== LOOP %d\n", i)
		logger := log.NewNopLogger()
		stateDB := dbm.NewMemDB()
		state, _ := sm.MakeGenesisStateFromFile(consensusReplayConfig.GenesisFile())
		privValidator := loadPrivValidator(consensusReplayConfig)
		blockDB := dbm.NewMemDB()
		cs := newConsensusStateWithConfigAndBlockStore(consensusReplayConfig, state, privValidator, kvstore.NewKVStoreApplication(), blockDB)
		cs.SetLogger(logger)
		ctx, cancel := context.WithCancel(context.Background())
		initFn(stateDB, cs, ctx)
		walFile := cs.config.WalFile()
		os.Remove(walFile)
		csWal, err := cs.OpenWAL(walFile)
		require.NoError(t, err)
		crashingWal.next = csWal
		crashingWal.msgIndex = 1
		cs.wal = crashingWal
		err = cs.Start()
		require.NoError(t, err)
		i++
		select {
		case err := <-walPanicked:
			t.Logf("WAL panicked: %v", err)
			startNewConsensusStateAndWaitForBlock(t, consensusReplayConfig, cs.Height, blockDB, stateDB)
			cs.Stop()
			cancel()
			if _, ok := err.(ReachedHeightToStopError); ok {
				break LOOP
			}
		case <-time.After(10 * time.Second):
			t.Fatal("WAL did not panic for 10 seconds (check the log)")
		}
	}
}
type crashingWAL struct {
	next         WAL
	panicCh      chan error
	heightToStop int64
	msgIndex                int  
	lastPanickedForMsgIndex int  
}
var _ WAL = &crashingWAL{}
type WALWriteError struct {
	msg string
}
func (e WALWriteError) Error() string {
	return e.msg
}
type ReachedHeightToStopError struct {
	height int64
}
func (e ReachedHeightToStopError) Error() string {
	return fmt.Sprintf("reached height to stop %d", e.height)
}
func (w *crashingWAL) Write(m WALMessage) {
	if endMsg, ok := m.(EndHeightMessage); ok {
		if endMsg.Height == w.heightToStop {
			w.panicCh <- ReachedHeightToStopError{endMsg.Height}
			runtime.Goexit()
		} else {
			w.next.Write(m)
		}
		return
	}
	if w.msgIndex > w.lastPanickedForMsgIndex {
		w.lastPanickedForMsgIndex = w.msgIndex
		_, file, line, _ := runtime.Caller(1)
		w.panicCh <- WALWriteError{fmt.Sprintf("failed to write %T to WAL (fileline: %s:%d)", m, file, line)}
		runtime.Goexit()
	} else {
		w.msgIndex++
		w.next.Write(m)
	}
}
func (w *crashingWAL) WriteSync(m WALMessage) {
	w.Write(m)
}
func (w *crashingWAL) FlushAndSync() error { return w.next.FlushAndSync() }
func (w *crashingWAL) SearchForEndHeight(height int64, options *WALSearchOptions) (rd io.ReadCloser, found bool, err error) {
	return w.next.SearchForEndHeight(height, options)
}
func (w *crashingWAL) Start() error { return w.next.Start() }
func (w *crashingWAL) Stop() error  { return w.next.Stop() }
func (w *crashingWAL) Wait()        { w.next.Wait() }
const (
	NUM_BLOCKS = 6
)
var (
	mempool = sm.MockMempool{}
	evpool  = sm.MockEvidencePool{}
)
var modes = []uint{0, 1, 2}
func TestHandshakeReplayAll(t *testing.T) {
	for i, m := range modes {
		config := ResetConfig(fmt.Sprintf("%s_%v", t.Name(), i))
		defer os.RemoveAll(config.RootDir)
		testHandshakeReplay(t, config, 0, m)
	}
}
func TestHandshakeReplaySome(t *testing.T) {
	for i, m := range modes {
		config := ResetConfig(fmt.Sprintf("%s_%v", t.Name(), i))
		defer os.RemoveAll(config.RootDir)
		testHandshakeReplay(t, config, 1, m)
	}
}
func TestHandshakeReplayOne(t *testing.T) {
	for i, m := range modes {
		config := ResetConfig(fmt.Sprintf("%s_%v", t.Name(), i))
		defer os.RemoveAll(config.RootDir)
		testHandshakeReplay(t, config, NUM_BLOCKS-1, m)
	}
}
func TestHandshakeReplayNone(t *testing.T) {
	for i, m := range modes {
		config := ResetConfig(fmt.Sprintf("%s_%v", t.Name(), i))
		defer os.RemoveAll(config.RootDir)
		testHandshakeReplay(t, config, NUM_BLOCKS, m)
	}
}
func tempWALWithData(data []byte) string {
	walFile, err := ioutil.TempFile("", "wal")
	if err != nil {
		panic(fmt.Errorf("failed to create temp WAL file: %v", err))
	}
	_, err = walFile.Write(data)
	if err != nil {
		panic(fmt.Errorf("failed to write to temp WAL file: %v", err))
	}
	if err := walFile.Close(); err != nil {
		panic(fmt.Errorf("failed to close temp WAL file: %v", err))
	}
	return walFile.Name()
}
func testHandshakeReplay(t *testing.T, config *cfg.Config, nBlocks int, mode uint) {
	walBody, err := WALWithNBlocks(t, NUM_BLOCKS)
	require.NoError(t, err)
	walFile := tempWALWithData(walBody)
	config.Consensus.SetWalFile(walFile)
	privVal := privval.LoadFilePV(config.PrivValidatorKeyFile(), config.PrivValidatorStateFile())
	wal, err := NewWAL(walFile)
	require.NoError(t, err)
	wal.SetLogger(log.TestingLogger())
	err = wal.Start()
	require.NoError(t, err)
	defer wal.Stop()
	chain, commits, err := makeBlockchainFromWAL(wal)
	require.NoError(t, err)
	stateDB, state, store := stateAndStore(config, privVal.GetPubKey(), kvstore.ProtocolVersion)
	store.chain = chain
	store.commits = commits
	state = buildTMStateFromChain(config, stateDB, state, chain, mode)
	latestAppHash := state.AppHash
	kvstoreApp := kvstore.NewPersistentKVStoreApplication(path.Join(config.DBDir(), "2"))
	clientCreator2 := proxy.NewLocalClientCreator(kvstoreApp)
	if nBlocks > 0 {
		proxyApp := proxy.NewAppConns(clientCreator2)
		stateDB, state, _ := stateAndStore(config, privVal.GetPubKey(), kvstore.ProtocolVersion)
		buildAppStateFromChain(proxyApp, stateDB, state, chain, nBlocks, mode)
	}
	genDoc, _ := sm.MakeGenesisDocFromFile(config.GenesisFile())
	handshaker := NewHandshaker(stateDB, state, store, genDoc)
	proxyApp := proxy.NewAppConns(clientCreator2)
	if err := proxyApp.Start(); err != nil {
		t.Fatalf("Error starting proxy app connections: %v", err)
	}
	defer proxyApp.Stop()
	if err := handshaker.Handshake(proxyApp); err != nil {
		t.Fatalf("Error on abci handshake: %v", err)
	}
	res, err := proxyApp.Query().InfoSync(abci.RequestInfo{Version: ""})
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(latestAppHash, res.LastBlockAppHash) {
		t.Fatalf("Expected app hashes to match after handshake/replay. got %X, expected %X", res.LastBlockAppHash, latestAppHash)
	}
	expectedBlocksToSync := NUM_BLOCKS - nBlocks
	if nBlocks == NUM_BLOCKS && mode > 0 {
		expectedBlocksToSync++
	} else if nBlocks > 0 && mode == 1 {
		expectedBlocksToSync++
	}
	if handshaker.NBlocks() != expectedBlocksToSync {
		t.Fatalf("Expected handshake to sync %d blocks, got %d", expectedBlocksToSync, handshaker.NBlocks())
	}
}
func applyBlock(stateDB dbm.DB, st sm.State, blk *types.Block, proxyApp proxy.AppConns) sm.State {
	testPartSize := types.BlockPartSizeBytes
	blockExec := sm.NewBlockExecutor(stateDB, log.TestingLogger(), proxyApp.Consensus(), mempool, evpool)
	blkID := types.BlockID{blk.Hash(), blk.MakePartSet(testPartSize).Header()}
	newState, err := blockExec.ApplyBlock(st, blkID, blk)
	if err != nil {
		panic(err)
	}
	return newState
}
func buildAppStateFromChain(proxyApp proxy.AppConns, stateDB dbm.DB,
	state sm.State, chain []*types.Block, nBlocks int, mode uint) {
	if err := proxyApp.Start(); err != nil {
		panic(err)
	}
	defer proxyApp.Stop()
	validators := types.TM2PB.ValidatorUpdates(state.Validators)
	if _, err := proxyApp.Consensus().InitChainSync(abci.RequestInitChain{
		Validators: validators,
	}); err != nil {
		panic(err)
	}
	switch mode {
	case 0:
		for i := 0; i < nBlocks; i++ {
			block := chain[i]
			state = applyBlock(stateDB, state, block, proxyApp)
		}
	case 1, 2:
		for i := 0; i < nBlocks-1; i++ {
			block := chain[i]
			state = applyBlock(stateDB, state, block, proxyApp)
		}
		if mode == 2 {
			state = applyBlock(stateDB, state, chain[nBlocks-1], proxyApp)
		}
	}
}
func buildTMStateFromChain(config *cfg.Config, stateDB dbm.DB, state sm.State, chain []*types.Block, mode uint) sm.State {
	clientCreator := proxy.NewLocalClientCreator(kvstore.NewPersistentKVStoreApplication(path.Join(config.DBDir(), "1")))
	proxyApp := proxy.NewAppConns(clientCreator)
	if err := proxyApp.Start(); err != nil {
		panic(err)
	}
	defer proxyApp.Stop()
	validators := types.TM2PB.ValidatorUpdates(state.Validators)
	if _, err := proxyApp.Consensus().InitChainSync(abci.RequestInitChain{
		Validators: validators,
	}); err != nil {
		panic(err)
	}
	switch mode {
	case 0:
		for _, block := range chain {
			state = applyBlock(stateDB, state, block, proxyApp)
		}
	case 1, 2:
		for _, block := range chain[:len(chain)-1] {
			state = applyBlock(stateDB, state, block, proxyApp)
		}
		applyBlock(stateDB, state, chain[len(chain)-1], proxyApp)
	}
	return state
}