diff --git a/consensus/replay.go b/consensus/replay.go index 02fa28c27..2dcd2e056 100644 --- a/consensus/replay.go +++ b/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 } } diff --git a/consensus/replay_file.go b/consensus/replay_file.go index 1182aaf04..fa7800b46 100644 --- a/consensus/replay_file.go +++ b/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 diff --git a/consensus/replay_test.go b/consensus/replay_test.go index 06a27d542..aa2b9a131 100644 --- a/consensus/replay_test.go +++ b/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 diff --git a/consensus/state.go b/consensus/state.go index 94ecd2a2c..6bb30e8d5 100644 --- a/consensus/state.go +++ b/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 diff --git a/consensus/test_data/build.sh b/consensus/test_data/build.sh index 1d6ef3c78..16cf96a60 100755 --- a/consensus/test_data/build.sh +++ b/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 } diff --git a/consensus/test_data/empty_block.cswal b/consensus/test_data/empty_block.cswal index 4d1ac1f8b..609f4ddf3 100644 Binary files a/consensus/test_data/empty_block.cswal and b/consensus/test_data/empty_block.cswal differ diff --git a/consensus/test_data/many_blocks.cswal b/consensus/test_data/many_blocks.cswal index c913bfb31..ab486b5a1 100644 Binary files a/consensus/test_data/many_blocks.cswal and b/consensus/test_data/many_blocks.cswal differ diff --git a/consensus/test_data/small_block1.cswal b/consensus/test_data/small_block1.cswal index c9f8d4bef..b7c7e777f 100644 Binary files a/consensus/test_data/small_block1.cswal and b/consensus/test_data/small_block1.cswal differ diff --git a/consensus/test_data/small_block2.cswal b/consensus/test_data/small_block2.cswal index 9901801e3..2ef077dcd 100644 Binary files a/consensus/test_data/small_block2.cswal and b/consensus/test_data/small_block2.cswal differ diff --git a/consensus/wal.go b/consensus/wal.go index 9144b09e8..51cb31193 100644 --- a/consensus/wal.go +++ b/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 } diff --git a/consensus/wal_test.go b/consensus/wal_test.go new file mode 100644 index 000000000..f032265c3 --- /dev/null +++ b/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) + } +} diff --git a/glide.lock b/glide.lock index a806e1f96..167ab77ca 100644 --- a/glide.lock +++ b/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 diff --git a/glide.yaml b/glide.yaml index a74870ab2..3fbcc5fc6 100644 --- a/glide.yaml +++ b/glide.yaml @@ -30,7 +30,7 @@ import: subpackages: - iavl - package: github.com/tendermint/tmlibs - version: develop + version: 21b2c26fb1b26edf5846792890e01eaa8a472508 subpackages: - autofile - cli diff --git a/scripts/cutWALUntil/main.go b/scripts/cutWALUntil/main.go new file mode 100644 index 000000000..e2886fad7 --- /dev/null +++ b/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)) + } + } +} diff --git a/scripts/wal2json/main.go b/scripts/wal2json/main.go new file mode 100644 index 000000000..46a3e17b0 --- /dev/null +++ b/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")) + } +} diff --git a/scripts/wal2json/wal2json b/scripts/wal2json/wal2json new file mode 100755 index 000000000..b7cab1467 Binary files /dev/null and b/scripts/wal2json/wal2json differ