Merge pull request #343 from tendermint/restart_test

Crash/Restart tests

consensus/replay.go

@@ -101,6 +101,7 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error {
 	// Search for height marker
 	gr, found, err = cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight))
 	if err == io.EOF {
+		log.Warn("Replay: wal.group.Search returned EOF", "height", csHeight)
 		return nil
 	} else if err != nil {
 		return err

consensus/state.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
 	"reflect"
 	"sync"
 	"time"
@@ -347,6 +348,23 @@ func (cs *ConsensusState) OnStart() error {
 		return err
 	}
 
+	// If the latest block was applied in the tmsp handshake,
+	// we may not have written the current height to the wal,
+	// so check here and write it if not found.
+	// TODO: remove this and run the handhsake/replay
+	// through the consensus state with a mock app
+	gr, found, err := cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(cs.Height))
+	if (err == io.EOF || !found) && cs.Step == RoundStepNewHeight {
+		log.Warn("Height not found in wal. Writing new height", "height", cs.Height)
+		rs := cs.RoundStateEvent()
+		cs.wal.Save(rs)
+	} else if err != nil {
+		return err
+	}
+	if gr != nil {
+		gr.Close()
+	}
+
 	// we need the timeoutRoutine for replay so
 	//  we don't block on the tick chan.
 	// NOTE: we will get a build up of garbage go routines

consensus/wal.go

@@ -104,4 +104,9 @@ func (wal *WAL) Save(wmsg WALMessage) {
 
 func (wal *WAL) writeHeight(height int) {
 	wal.group.WriteLine(Fmt("#HEIGHT: %v", height))
+
+	// TODO: only flush when necessary
+	if err := wal.group.Flush(); err != nil {
+		PanicQ(Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
+	}
 }