binary format for WAL

consensus/replay.go

@@ -12,7 +12,6 @@ import (
 	"time"
 
 	abci "github.com/tendermint/abci/types"
-	wire "github.com/tendermint/go-wire"
 	auto "github.com/tendermint/tmlibs/autofile"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"
@@ -38,29 +37,11 @@ var crc32c = crc32.MakeTable(crc32.Castagnoli)
 // as if it were received in receiveRoutine
 // Lines that start with "#" are ignored.
 // NOTE: receiveRoutine should not be running
-func (cs *ConsensusState) readReplayMessage(msgBytes []byte, newStepCh chan interface{}) error {
-	// Skip over empty and meta lines
-	if len(msgBytes) == 0 || msgBytes[0] == '#' {
+func (cs *ConsensusState) readReplayMessage(msg *TimedWALMessage, newStepCh chan interface{}) error {
+	// Skip over meta lines
+	if _, ok := msg.Msg.(EndHeightMessage); ok {
 		return nil
 	}
-	var err error
-	var msg TimedWALMessage
-	wire.ReadJSON(&msg, msgBytes, &err)
-	if err != nil {
-		fmt.Println("MsgBytes:", msgBytes, string(msgBytes))
-		return fmt.Errorf("Error reading json data: %v", err)
-	}
-	// check checksum
-	innerMsgBytes := wire.JSONBytes(msg.Msg)
-	crc := crc32.Checksum(innerMsgBytes, crc32c)
-	if crc != msg.CRC {
-		return fmt.Errorf("Checksums do not match. Original: %v, but calculated: %v", msg.CRC, crc)
-	}
-	// check msg size (optional)
-	msgSize := uint32(len(innerMsgBytes))
-	if msgSize != msg.MsgSize {
-		return fmt.Errorf("Sizes do not match. Original: %v, but calculated: %v", msg.MsgSize, msgSize)
-	}
 
 	// for logging
 	switch m := msg.Msg.(type) {
@@ -118,7 +99,7 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error {
 	// Ensure that ENDHEIGHT for this height doesn't exist
 	// NOTE: This is just a sanity check. As far as we know things work fine without it,
 	// and Handshake could reuse ConsensusState if it weren't for this check (since we can crash after writing ENDHEIGHT).
-	gr, found, err := cs.wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(csHeight))
+	gr, found, err := cs.wal.SearchForEndHeight(uint64(csHeight))
 	if gr != nil {
 		gr.Close()
 	}
@@ -127,7 +108,7 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error {
 	}
 
 	// Search for last height marker
-	gr, found, err = cs.wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(csHeight-1))
+	gr, found, err = cs.wal.SearchForEndHeight(uint64(csHeight - 1))
 	if err == io.EOF {
 		cs.Logger.Error("Replay: wal.group.Search returned EOF", "#ENDHEIGHT", csHeight-1)
 	} else if err != nil {
@@ -141,19 +122,21 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error {
 
 	cs.Logger.Info("Catchup by replaying consensus messages", "height", csHeight)
 
+	var msg *TimedWALMessage
+	dec := WALDecoder{gr}
+
 	for {
-		line, err := gr.ReadLine()
+		msg, err = dec.Decode()
+		if err == io.EOF {
+			break
+		}
 		if err != nil {
-			if err == io.EOF {
-				break
-			} else {
-				return err
-			}
+			return err
 		}
 		// NOTE: since the priv key is set when the msgs are received
 		// it will attempt to eg double sign but we can just ignore it
 		// since the votes will be replayed and we'll get to the next step
-		if err := cs.readReplayMessage([]byte(line), nil); err != nil {
+		if err := cs.readReplayMessage(msg, nil); err != nil {
 			return err
 		}
 	}

consensus/replay_file.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"strconv"
 	"strings"
@@ -53,12 +54,20 @@ func (cs *ConsensusState) ReplayFile(file string, console bool) error {
 	defer pb.fp.Close()
 
 	var nextN int // apply N msgs in a row
-	for pb.scanner.Scan() {
+	var msg *TimedWALMessage
+	for {
 		if nextN == 0 && console {
 			nextN = pb.replayConsoleLoop()
 		}
 
-		if err := pb.cs.readReplayMessage(pb.scanner.Bytes(), newStepCh); err != nil {
+		msg, err = pb.dec.Decode()
+		if err == io.EOF {
+			return nil
+		} else {
+			return err
+		}
+
+		if err := pb.cs.readReplayMessage(msg, newStepCh); err != nil {
 			return err
 		}
 
@@ -76,9 +85,9 @@ func (cs *ConsensusState) ReplayFile(file string, console bool) error {
 type playback struct {
 	cs *ConsensusState
 
-	fp      *os.File
-	scanner *bufio.Scanner
-	count   int // how many lines/msgs into the file are we
+	fp    *os.File
+	dec   *WALDecoder
+	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
@@ -91,7 +100,7 @@ func newPlayback(fileName string, fp *os.File, cs *ConsensusState, genState *sm.
 		fp:           fp,
 		fileName:     fileName,
 		genesisState: genState,
-		scanner:      bufio.NewScanner(fp),
+		dec:          NewWALDecoder(fp),
 	}
 }
 
@@ -111,13 +120,20 @@ func (pb *playback) replayReset(count int, newStepCh chan interface{}) error {
 		return err
 	}
 	pb.fp = fp
-	pb.scanner = bufio.NewScanner(fp)
+	pb.dec = NewWALDecoder(fp)
 	count = pb.count - count
 	fmt.Printf("Reseting from %d to %d\n", pb.count, count)
 	pb.count = 0
 	pb.cs = newCS
-	for i := 0; pb.scanner.Scan() && i < count; i++ {
-		if err := pb.cs.readReplayMessage(pb.scanner.Bytes(), newStepCh); err != nil {
+	var msg *TimedWALMessage
+	for i := 0; i < count; i++ {
+		msg, err = pb.dec.Decode()
+		if err == io.EOF {
+			return nil
+		} else {
+			return err
+		}
+		if err := pb.cs.readReplayMessage(msg, newStepCh); err != nil {
 			return err
 		}
 		pb.count += 1

consensus/replay_test.go

@@ -8,7 +8,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
-	"strings"
 	"testing"
 	"time"
 
@@ -58,14 +57,14 @@ var baseStepChanges = []int{3, 6, 8}
 
 // test recovery from each line in each testCase
 var testCases = []*testCase{
-	newTestCase("empty_block", baseStepChanges),  // empty block (has 1 block part)
-	newTestCase("small_block1", baseStepChanges), // small block with txs in 1 block part
-	newTestCase("small_block2", []int{3, 7, 9}),  // small block with txs across 6 smaller block parts
+	newTestCase("empty_block", baseStepChanges),   // empty block (has 1 block part)
+	newTestCase("small_block1", baseStepChanges),  // small block with txs in 1 block part
+	newTestCase("small_block2", []int{3, 12, 14}), // small block with txs across 6 smaller block parts
 }
 
 type testCase struct {
 	name    string
-	log     string       //full cs wal
+	log     []byte       //full cs wal
 	stepMap map[int]int8 // map lines of log to privval step
 
 	proposeLine   int
@@ -100,29 +99,27 @@ func newMapFromChanges(changes []int) map[int]int8 {
 	return m
 }
 
-func readWAL(p string) string {
+func readWAL(p string) []byte {
 	b, err := ioutil.ReadFile(p)
 	if err != nil {
 		panic(err)
 	}
-	return string(b)
+	return b
 }
 
-func writeWAL(walMsgs string) string {
-	tempDir := os.TempDir()
-	walDir := path.Join(tempDir, "/wal"+cmn.RandStr(12))
-	walFile := path.Join(walDir, "wal")
-	// Create WAL directory
-	err := cmn.EnsureDir(walDir, 0700)
+func writeWAL(walMsgs []byte) string {
+	walFile, err := ioutil.TempFile("", "wal")
 	if err != nil {
-		panic(err)
+		panic(fmt.Errorf("failed to create temp WAL file: %v", err))
 	}
-	// Write the needed WAL to file
-	err = cmn.WriteFile(walFile, []byte(walMsgs), 0600)
+	_, err = walFile.Write(walMsgs)
 	if err != nil {
-		panic(err)
+		panic(fmt.Errorf("failed to write to temp WAL file: %v", err))
 	}
-	return walFile
+	if err := walFile.Close(); err != nil {
+		panic(fmt.Errorf("failed to close temp WAL file: %v", err))
+	}
+	return walFile.Name()
 }
 
 func waitForBlock(newBlockCh chan interface{}, thisCase *testCase, i int) {
@@ -167,7 +164,7 @@ func toPV(pv types.PrivValidator) *types.PrivValidatorFS {
 	return pv.(*types.PrivValidatorFS)
 }
 
-func setupReplayTest(t *testing.T, thisCase *testCase, nLines int, crashAfter bool) (*ConsensusState, chan interface{}, string, string) {
+func setupReplayTest(t *testing.T, thisCase *testCase, nLines int, crashAfter bool) (*ConsensusState, chan interface{}, []byte, string) {
 	t.Log("-------------------------------------")
 	t.Logf("Starting replay test %v (of %d lines of WAL). Crash after = %v", thisCase.name, nLines, crashAfter)
 
@@ -176,11 +173,13 @@ func setupReplayTest(t *testing.T, thisCase *testCase, nLines int, crashAfter bo
 		lineStep -= 1
 	}
 
-	split := strings.Split(thisCase.log, "\n")
+	split := bytes.Split(thisCase.log, walSeparator)
 	lastMsg := split[nLines]
 
 	// we write those lines up to (not including) one with the signature
-	walFile := writeWAL(strings.Join(split[:nLines], "\n") + "\n")
+	bytes := bytes.Join(split[:nLines], walSeparator)
+	bytes = append(bytes, walSeparator...)
+	walFile := writeWAL(bytes)
 
 	cs := fixedConsensusStateDummy()
 
@@ -195,14 +194,18 @@ func setupReplayTest(t *testing.T, thisCase *testCase, nLines int, crashAfter bo
 	return cs, newBlockCh, lastMsg, walFile
 }
 
-func readTimedWALMessage(t *testing.T, walMsg string) TimedWALMessage {
-	var err error
-	var msg TimedWALMessage
-	wire.ReadJSON(&msg, []byte(walMsg), &err)
+func readTimedWALMessage(t *testing.T, rawMsg []byte) TimedWALMessage {
+	b := bytes.NewBuffer(rawMsg)
+	_, err := b.Write(walSeparator)
+	if err != nil {
+		t.Fatal(err)
+	}
+	dec := NewWALDecoder(b)
+	msg, err := dec.Decode()
 	if err != nil {
 		t.Fatalf("Error reading json data: %v", err)
 	}
-	return msg
+	return *msg
 }
 
 //-----------------------------------------------
@@ -211,10 +214,14 @@ func readTimedWALMessage(t *testing.T, walMsg string) TimedWALMessage {
 
 func TestWALCrashAfterWrite(t *testing.T) {
 	for _, thisCase := range testCases {
-		split := strings.Split(thisCase.log, "\n")
-		for i := 0; i < len(split)-1; i++ {
-			cs, newBlockCh, _, walFile := setupReplayTest(t, thisCase, i+1, true)
-			runReplayTest(t, cs, walFile, newBlockCh, thisCase, i+1)
+		splitSize := bytes.Count(thisCase.log, walSeparator)
+		for i := 0; i < splitSize-1; i++ {
+			t.Run(fmt.Sprintf("%s:%d", thisCase.name, i), func(t *testing.T) {
+				cs, newBlockCh, _, walFile := setupReplayTest(t, thisCase, i+1, true)
+				runReplayTest(t, cs, walFile, newBlockCh, thisCase, i+1)
+				// cleanup
+				os.Remove(walFile)
+			})
 		}
 	}
 }
@@ -226,14 +233,18 @@ func TestWALCrashAfterWrite(t *testing.T) {
 func TestWALCrashBeforeWritePropose(t *testing.T) {
 	for _, thisCase := range testCases {
 		lineNum := thisCase.proposeLine
-		// setup replay test where last message is a proposal
-		cs, newBlockCh, proposalMsg, walFile := setupReplayTest(t, thisCase, lineNum, false)
-		msg := readTimedWALMessage(t, proposalMsg)
-		proposal := msg.Msg.(msgInfo).Msg.(*ProposalMessage)
-		// Set LastSig
-		toPV(cs.privValidator).LastSignBytes = types.SignBytes(cs.state.ChainID, proposal.Proposal)
-		toPV(cs.privValidator).LastSignature = proposal.Proposal.Signature
-		runReplayTest(t, cs, walFile, newBlockCh, thisCase, lineNum)
+		t.Run(fmt.Sprintf("%s:%d", thisCase.name, lineNum), func(t *testing.T) {
+			// setup replay test where last message is a proposal
+			cs, newBlockCh, proposalMsg, walFile := setupReplayTest(t, thisCase, lineNum, false)
+			msg := readTimedWALMessage(t, proposalMsg)
+			proposal := msg.Msg.(msgInfo).Msg.(*ProposalMessage)
+			// Set LastSig
+			toPV(cs.privValidator).LastSignBytes = types.SignBytes(cs.state.ChainID, proposal.Proposal)
+			toPV(cs.privValidator).LastSignature = proposal.Proposal.Signature
+			runReplayTest(t, cs, walFile, newBlockCh, thisCase, lineNum)
+			// cleanup
+			os.Remove(walFile)
+		})
 	}
 }
 
@@ -315,7 +326,7 @@ func testHandshakeReplay(t *testing.T, nBlocks int, mode uint) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	walFile := writeWAL(string(walBody))
+	walFile := writeWAL(walBody)
 	config.Consensus.SetWalFile(walFile)
 
 	privVal := types.LoadPrivValidatorFS(config.PrivValidatorFile())
@@ -465,7 +476,7 @@ func buildTMStateFromChain(config *cfg.Config, state *sm.State, chain []*types.B
 
 func makeBlockchainFromWAL(wal *WAL) ([]*types.Block, []*types.Commit, error) {
 	// Search for height marker
-	gr, found, err := wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(0))
+	gr, found, err := wal.SearchForEndHeight(0)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -479,20 +490,18 @@ func makeBlockchainFromWAL(wal *WAL) ([]*types.Block, []*types.Commit, error) {
 	var blockParts *types.PartSet
 	var blocks []*types.Block
 	var commits []*types.Commit
+
+	dec := NewWALDecoder(gr)
 	for {
-		line, err := gr.ReadLine()
-		if err != nil {
-			if err == io.EOF {
-				break
-			} else {
-				return nil, nil, err
-			}
+		msg, err := dec.Decode()
+		if err == io.EOF {
+			break
 		}
-
-		piece, err := readPieceFromWAL([]byte(line))
 		if err != nil {
 			return nil, nil, err
 		}
+
+		piece := readPieceFromWAL(msg)
 		if piece == nil {
 			continue
 		}
@@ -528,17 +537,10 @@ func makeBlockchainFromWAL(wal *WAL) ([]*types.Block, []*types.Commit, error) {
 	return blocks, commits, nil
 }
 
-func readPieceFromWAL(msgBytes []byte) (interface{}, error) {
-	// Skip over empty and meta lines
-	if len(msgBytes) == 0 || msgBytes[0] == '#' {
-		return nil, nil
-	}
-	var err error
-	var msg TimedWALMessage
-	wire.ReadJSON(&msg, msgBytes, &err)
-	if err != nil {
-		fmt.Println("MsgBytes:", msgBytes, string(msgBytes))
-		return nil, fmt.Errorf("Error reading json data: %v", err)
+func readPieceFromWAL(msg *TimedWALMessage) interface{} {
+	// Skip meta lines
+	if _, ok := msg.Msg.(EndHeightMessage); ok {
+		return nil
 	}
 
 	// for logging
@@ -546,14 +548,15 @@ func readPieceFromWAL(msgBytes []byte) (interface{}, error) {
 	case msgInfo:
 		switch msg := m.Msg.(type) {
 		case *ProposalMessage:
-			return &msg.Proposal.BlockPartsHeader, nil
+			return &msg.Proposal.BlockPartsHeader
 		case *BlockPartMessage:
-			return msg.Part, nil
+			return msg.Part
 		case *VoteMessage:
-			return msg.Vote, nil
+			return msg.Vote
 		}
 	}
-	return nil, nil
+
+	return nil
 }
 
 // fresh state and mock store

consensus/state.go

@@ -1188,7 +1188,7 @@ func (cs *ConsensusState) finalizeCommit(height int) {
 	// As is, ConsensusState should not be started again
 	// until we successfully call ApplyBlock (ie. here or in Handshake after restart)
 	if cs.wal != nil {
-		cs.wal.writeEndHeight(height)
+		cs.wal.Save(EndHeightMessage{uint64(height)})
 	}
 
 	fail.Fail() // XXX

consensus/test_data/build.sh

@@ -16,6 +16,11 @@ if ! hash tendermint 2>/dev/null; then
     make install
 fi
 
+# Make sure we have a cutWALUntil binary.
+if ! hash ./scripts/cutWALUntil/cutWALUntil 2>/dev/null; then
+	cd ./scripts/cutWALUntil/ && go build && cd - || exit
+fi
+
 # specify a dir to copy
 # TODO: eventually we should replace with `tendermint init --test`
 DIR_TO_COPY=$HOME/.tendermint_test/consensus_state_test
@@ -39,10 +44,8 @@ function empty_block(){
 	sleep 5
 	killall tendermint
 
-	# /q would print up to and including the match, then quit.
-	# /Q doesn't include the match.
-	# http://unix.stackexchange.com/questions/11305/grep-show-all-the-file-up-to-the-match
-	sed -e "/ENDHEIGHT: 1/Q" ~/.tendermint/data/cs.wal/wal > consensus/test_data/empty_block.cswal
+	./scripts/cutWALUntil/cutWALUntil ~/.tendermint/data/cs.wal/wal 1 consensus/test_data/new_empty_block.cswal
+	mv consensus/test_data/new_empty_block.cswal consensus/test_data/empty_block.cswal
 
 	reset
 }
@@ -56,7 +59,8 @@ function many_blocks(){
 	killall tendermint
 	kill -9 $PID
 
-	sed -e '/ENDHEIGHT: 6/Q' ~/.tendermint/data/cs.wal/wal  > consensus/test_data/many_blocks.cswal
+	./scripts/cutWALUntil/cutWALUntil ~/.tendermint/data/cs.wal/wal 6 consensus/test_data/new_many_blocks.cswal
+	mv consensus/test_data/new_many_blocks.cswal consensus/test_data/many_blocks.cswal
 
 	reset
 }
@@ -71,7 +75,8 @@ function small_block1(){
 	killall tendermint
 	kill -9 $PID
 
-	sed -e '/ENDHEIGHT: 1/Q' ~/.tendermint/data/cs.wal/wal  > consensus/test_data/small_block1.cswal
+	./scripts/cutWALUntil/cutWALUntil ~/.tendermint/data/cs.wal/wal 1 consensus/test_data/new_small_block1.cswal
+	mv consensus/test_data/new_small_block1.cswal consensus/test_data/small_block1.cswal
 
 	reset
 }
@@ -79,8 +84,8 @@ function small_block1(){
 
 # small block 2 (part size = 64)
 function small_block2(){
-	cat ~/.tendermint/genesis.json | jq  '. + {"consensus_params": {"block_size_params": {"max_bytes":1000000}, "block_gossip_params": {"block_part_size_bytes":512}}}' > genesis.json.new
-	mv genesis.json.new ~/.tendermint/genesis.json
+	cat ~/.tendermint/genesis.json | jq '. + {consensus_params: {block_size_params: {max_bytes: 22020096}, block_gossip_params: {block_part_size_bytes: 512}}}' > ~/.tendermint/new_genesis.json
+	mv ~/.tendermint/new_genesis.json ~/.tendermint/genesis.json
 	bash scripts/txs/random.sh 1000 36657 &> /dev/null &
 	PID=$!
 	tendermint node --proxy_app=persistent_dummy &> /dev/null &
@@ -88,7 +93,8 @@ function small_block2(){
 	killall tendermint
 	kill -9 $PID
 
-	sed -e '/ENDHEIGHT: 1/Q' ~/.tendermint/data/cs.wal/wal  > consensus/test_data/small_block2.cswal
+	./scripts/cutWALUntil/cutWALUntil ~/.tendermint/data/cs.wal/wal 1 consensus/test_data/new_small_block2.cswal
+	mv consensus/test_data/new_small_block2.cswal consensus/test_data/small_block2.cswal
 
 	reset
 }

consensus/wal.go

@@ -1,7 +1,11 @@
 package consensus
 
 import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
 	"hash/crc32"
+	"io"
 	"time"
 
 	wire "github.com/tendermint/go-wire"
@@ -13,11 +17,18 @@ import (
 //--------------------------------------------------------
 // types and functions for savings consensus messages
 
+var (
+	walSeparator = []byte{55, 127, 6, 130} // 0x377f0682 - magic number (can only affect tests)
+)
+
 type TimedWALMessage struct {
-	Time    time.Time  `json:"time"` // for debugging purposes
-	CRC     uint32     `json:"crc"`
-	MsgSize uint32     `json:"msg_size"`
-	Msg     WALMessage `json:"msg"`
+	Time time.Time  `json:"time"` // for debugging purposes
+	Msg  WALMessage `json:"msg"`
+}
+
+// EndHeightMessage marks the end of the given height inside WAL.
+type EndHeightMessage struct {
+	Height uint64 `json:"height"`
 }
 
 type WALMessage interface{}
@@ -27,6 +38,7 @@ var _ = wire.RegisterInterface(
 	wire.ConcreteType{types.EventDataRoundState{}, 0x01},
 	wire.ConcreteType{msgInfo{}, 0x02},
 	wire.ConcreteType{timeoutInfo{}, 0x03},
+	wire.ConcreteType{EndHeightMessage{}, 0x04},
 )
 
 //--------------------------------------------------------
@@ -41,6 +53,8 @@ type WAL struct {
 
 	group *auto.Group
 	light bool // ignore block parts
+
+	enc *WALEncoder
 }
 
 func NewWAL(walFile string, light bool) (*WAL, error) {
@@ -51,6 +65,7 @@ func NewWAL(walFile string, light bool) (*WAL, error) {
 	wal := &WAL{
 		group: group,
 		light: light,
+		enc:   NewWALEncoder(group),
 	}
 	wal.BaseService = *cmn.NewBaseService(nil, "WAL", wal)
 	return wal, nil
@@ -61,7 +76,7 @@ func (wal *WAL) OnStart() error {
 	if err != nil {
 		return err
 	} else if size == 0 {
-		wal.writeEndHeight(0)
+		wal.Save(EndHeightMessage{0})
 	}
 	_, err = wal.group.Start()
 	return err
@@ -73,38 +88,195 @@ func (wal *WAL) OnStop() {
 }
 
 // called in newStep and for each pass in receiveRoutine
-func (wal *WAL) Save(wmsg WALMessage) {
+func (wal *WAL) Save(msg WALMessage) {
 	if wal == nil {
 		return
 	}
+
 	if wal.light {
 		// in light mode we only write new steps, timeouts, and our own votes (no proposals, block parts)
-		if mi, ok := wmsg.(msgInfo); ok {
+		if mi, ok := msg.(msgInfo); ok {
 			if mi.PeerKey != "" {
 				return
 			}
 		}
 	}
+
 	// Write the wal message
-	innerMsgBytes := wire.JSONBytes(wmsg)
-	crc := crc32.Checksum(innerMsgBytes, crc32c)
-	wmsgSize := uint32(len(innerMsgBytes))
-	var wmsgBytes = wire.JSONBytes(TimedWALMessage{time.Now(), crc, wmsgSize, wmsg})
-	err := wal.group.WriteLine(string(wmsgBytes))
-	if err != nil {
-		cmn.PanicQ(cmn.Fmt("Error writing msg to consensus wal. Error: %v \n\nMessage: %v", err, wmsg))
+	if err := wal.enc.Encode(&TimedWALMessage{time.Now(), msg}); err != nil {
+		cmn.PanicQ(cmn.Fmt("Error writing msg to consensus wal: %v \n\nMessage: %v", err, msg))
 	}
+
 	// TODO: only flush when necessary
 	if err := wal.group.Flush(); err != nil {
 		cmn.PanicQ(cmn.Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
 	}
 }
 
-func (wal *WAL) writeEndHeight(height int) {
-	wal.group.WriteLine(cmn.Fmt("#ENDHEIGHT: %v", height))
+// SearchForEndHeight searches for the EndHeightMessage with the height and
+// returns an auto.GroupReader, whenever it was found or not and an error.
+// Group reader will be nil if found equals false.
+//
+// CONTRACT: caller must close group reader.
+func (wal *WAL) SearchForEndHeight(height uint64) (gr *auto.GroupReader, found bool, err error) {
+	var msg *TimedWALMessage
 
-	// TODO: only flush when necessary
-	if err := wal.group.Flush(); err != nil {
-		cmn.PanicQ(cmn.Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
+	// NOTE: starting from the last file in the group because we're usually
+	// searching for the last height. See replay.go
+	min, max := wal.group.MinIndex(), wal.group.MaxIndex()
+	wal.Logger.Debug("Searching for height", "height", height, "min", min, "max", max)
+	for index := max; index >= min; index-- {
+		gr, err = wal.group.NewReader(index)
+		if err != nil {
+			return nil, false, err
+		}
+
+		dec := NewWALDecoder(gr)
+		for {
+			msg, err = dec.Decode()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				gr.Close()
+				return nil, false, err
+			}
+
+			if m, ok := msg.Msg.(EndHeightMessage); ok {
+				if m.Height == height { // found
+					wal.Logger.Debug("Found", "height", height, "index", index)
+					return gr, true, nil
+				} else if m.Height < height {
+					// we will never find it because we're starting from the end
+					gr.Close()
+					return nil, false, nil
+				}
+			}
+		}
+
+		gr.Close()
+	}
+
+	return nil, false, nil
+}
+
+///////////////////////////////////////////////////////////////////////////////
+
+// A WALEncoder writes custom-encoded WAL messages to an output stream.
+//
+// Format: 4 bytes CRC sum + 4 bytes length + arbitrary-length value (go-wire encoded)
+type WALEncoder struct {
+	wr io.Writer
+}
+
+// NewWALEncoder returns a new encoder that writes to wr.
+func NewWALEncoder(wr io.Writer) *WALEncoder {
+	return &WALEncoder{wr}
+}
+
+// Encode writes the custom encoding of v to the stream.
+func (enc *WALEncoder) Encode(v interface{}) error {
+	data := wire.BinaryBytes(v)
+
+	crc := crc32.Checksum(data, crc32c)
+	length := uint32(len(data))
+	totalLength := 8 + int(length)
+
+	msg := make([]byte, totalLength)
+	binary.BigEndian.PutUint32(msg[0:4], crc)
+	binary.BigEndian.PutUint32(msg[4:8], length)
+	copy(msg[8:], data)
+
+	_, err := enc.wr.Write(msg)
+
+	if err == nil {
+		// TODO [Anton Kaliaev 23 Oct 2017]: remove separator
+		_, err = enc.wr.Write(walSeparator)
+	}
+
+	return err
+}
+
+///////////////////////////////////////////////////////////////////////////////
+
+// A WALDecoder reads and decodes custom-encoded WAL messages from an input
+// stream. See WALEncoder for the format used.
+//
+// It will also compare the checksums and make sure data size is equal to the
+// length from the header. If that is not the case, error will be returned.
+type WALDecoder struct {
+	rd io.Reader
+}
+
+// NewWALDecoder returns a new decoder that reads from rd.
+func NewWALDecoder(rd io.Reader) *WALDecoder {
+	return &WALDecoder{rd}
+}
+
+// Decode reads the next custom-encoded value from its input and stores it in
+// the value pointed to by v.
+func (dec *WALDecoder) Decode() (*TimedWALMessage, error) {
+	b := make([]byte, 4)
+
+	n, err := dec.rd.Read(b)
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to read checksum: %v", err)
+	}
+	crc := binary.BigEndian.Uint32(b)
+
+	b = make([]byte, 4)
+	n, err = dec.rd.Read(b)
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to read length: %v", err)
+	}
+	length := binary.BigEndian.Uint32(b)
+
+	data := make([]byte, length)
+	n, err = dec.rd.Read(data)
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		return nil, fmt.Errorf("not enough bytes for data: %v (want: %d, read: %v)", err, length, n)
+	}
+
+	// check checksum before decoding data
+	actualCRC := crc32.Checksum(data, crc32c)
+	if actualCRC != crc {
+		return nil, fmt.Errorf("checksums do not match: (read: %v, actual: %v)", crc, actualCRC)
+	}
+
+	var nn int
+	var res *TimedWALMessage
+	res = wire.ReadBinary(&TimedWALMessage{}, bytes.NewBuffer(data), int(length), &nn, &err).(*TimedWALMessage)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode data: %v", err)
+	}
+
+	// TODO [Anton Kaliaev 23 Oct 2017]: remove separator
+	if err = readSeparator(dec.rd); err != nil {
+		return nil, err
+	}
+
+	return res, err
+}
+
+// readSeparator reads a separator from r. It returns any error from underlying
+// reader or if it's not a separator.
+func readSeparator(r io.Reader) error {
+	b := make([]byte, len(walSeparator))
+	_, err := r.Read(b)
+	if err != nil {
+		return fmt.Errorf("failed to read separator: %v", err)
+	}
+	if !bytes.Equal(b, walSeparator) {
+		return fmt.Errorf("not a separator: %v", b)
 	}
+	return nil
 }

consensus/wal_test.go

@@ -0,0 +1,37 @@
+package consensus
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"github.com/tendermint/tendermint/consensus/types"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWALEncoderDecoder(t *testing.T) {
+	now := time.Now()
+	msgs := []TimedWALMessage{
+		TimedWALMessage{Time: now, Msg: EndHeightMessage{0}},
+		TimedWALMessage{Time: now, Msg: timeoutInfo{Duration: time.Second, Height: 1, Round: 1, Step: types.RoundStepPropose}},
+	}
+
+	b := new(bytes.Buffer)
+
+	for _, msg := range msgs {
+		b.Reset()
+
+		enc := NewWALEncoder(b)
+		err := enc.Encode(&msg)
+		require.NoError(t, err)
+
+		dec := NewWALDecoder(b)
+		decoded, err := dec.Decode()
+		require.NoError(t, err)
+
+		assert.Equal(t, msg.Time.Truncate(time.Millisecond), decoded.Time)
+		assert.Equal(t, msg.Msg, decoded.Msg)
+	}
+}

glide.lock

@@ -1,16 +1,18 @@
-hash: 9867fa4543ca4daea1a96a3883a7f483819c067ca34ed6d3aa67aace4a289e93
-updated: 2017-10-23T10:01:08.326324082-04:00
+hash: f012b05ea79172925b7f896b475d2684a948ac94efaf28d55de8397b673a89e3
+updated: 2017-10-23T18:19:06.286350807Z
 imports:
 - name: github.com/btcsuite/btcd
-  version: c7588cbf7690cd9f047a28efa2dcd8f2435a4e5e
+  version: b8df516b4b267acf2de46be593a9d948d1d2c420
   subpackages:
   - btcec
+- name: github.com/btcsuite/fastsha256
+  version: 637e656429416087660c84436a2a035d69d54e2e
 - name: github.com/ebuchman/fail-test
   version: 95f809107225be108efcf10a3509e4ea6ceef3c4
 - name: github.com/fsnotify/fsnotify
   version: 4da3e2cfbabc9f751898f250b49f2439785783a1
 - name: github.com/go-kit/kit
-  version: e2b298466b32c7cd5579a9b9b07e968fc9d9452c
+  version: d67bb4c202e3b91377d1079b110a6c9ce23ab2f8
   subpackages:
   - log
   - log/level
@@ -122,7 +124,7 @@ imports:
   subpackages:
   - iavl
 - name: github.com/tendermint/tmlibs
-  version: 8e5266a9ef2527e68a1571f932db8228a331b556
+  version: 21b2c26fb1b26edf5846792890e01eaa8a472508
   subpackages:
   - autofile
   - cli

glide.yaml

@@ -30,7 +30,7 @@ import:
   subpackages:
   - iavl
 - package: github.com/tendermint/tmlibs
-  version: develop
+  version: 21b2c26fb1b26edf5846792890e01eaa8a472508
   subpackages:
   - autofile
   - cli

scripts/cutWALUntil/main.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"strconv"
+
+	cs "github.com/tendermint/tendermint/consensus"
+)
+
+func main() {
+	var heightToStop uint64
+	var err error
+	if heightToStop, err = strconv.ParseUint(os.Args[2], 10, 64); err != nil {
+		panic(fmt.Errorf("failed to parse height: %v (format: cutWALUntil in heightToStop out)", err))
+	}
+
+	in, err := os.Open(os.Args[1])
+	if err != nil {
+		panic(fmt.Errorf("failed to open WAL file: %v (format: cutWALUntil in heightToStop out)", err))
+	}
+	defer in.Close()
+
+	out, err := os.Create(os.Args[3])
+	if err != nil {
+		panic(fmt.Errorf("failed to open WAL file: %v (format: cutWALUntil in heightToStop out)", err))
+	}
+	defer out.Close()
+
+	enc := cs.NewWALEncoder(out)
+
+	dec := cs.NewWALDecoder(in)
+	for {
+		msg, err := dec.Decode()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			panic(fmt.Errorf("failed to decode msg: %v", err))
+		}
+
+		if m, ok := msg.Msg.(cs.EndHeightMessage); ok {
+			if m.Height == heightToStop {
+				break
+			}
+		}
+
+		err = enc.Encode(msg)
+		if err != nil {
+			panic(fmt.Errorf("failed to encode msg: %v", err))
+		}
+	}
+}

scripts/wal2json/main.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+
+	cs "github.com/tendermint/tendermint/consensus"
+)
+
+func main() {
+	f, err := os.Open(os.Args[1])
+	if err != nil {
+		panic(fmt.Errorf("failed to open WAL file: %v", err))
+	}
+	defer f.Close()
+
+	dec := cs.NewWALDecoder(f)
+	for {
+		msg, err := dec.Decode()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			panic(fmt.Errorf("failed to decode msg: %v", err))
+		}
+
+		json, err := json.Marshal(msg)
+		if err != nil {
+			panic(fmt.Errorf("failed to marshal msg: %v", err))
+		}
+
+		os.Stdout.Write(json)
+		os.Stdout.Write([]byte("\n"))
+	}
+}