diff --git a/consensus/replay.go b/consensus/replay.go index 53f876084..0adb333c7 100644 --- a/consensus/replay.go +++ b/consensus/replay.go @@ -105,7 +105,7 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error { // 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)) if found { - return errors.New(Fmt("WAL should not contain height %d.", csHeight)) + return errors.New(Fmt("WAL should not contain #ENDHEIGHT %d.", csHeight)) } if gr != nil { gr.Close() @@ -114,13 +114,13 @@ func (cs *ConsensusState) catchupReplay(csHeight int) error { // Search for last height marker gr, found, err = cs.wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(csHeight-1)) if err == io.EOF { - log.Warn("Replay: wal.group.Search returned EOF", "height", csHeight-1) + log.Warn("Replay: wal.group.Search returned EOF", "#ENDHEIGHT", csHeight-1) return nil } else if err != nil { return err } if !found { - return errors.New(Fmt("WAL does not contain height %d.", csHeight)) + return errors.New(Fmt("Cannot replay height %d. WAL does not contain #ENDHEIGHT for %d.", csHeight, csHeight-1)) } defer gr.Close() @@ -275,18 +275,18 @@ func (h *Handshaker) ReplayBlocks(appHash []byte, appBlockHeight int, proxyApp p } else if appBlockHeight == stateBlockHeight { // We haven't run Commit (both the state and app are one block behind), - // so ApplyBlock with the real app. + // so replayBlock with the real app. // NOTE: We could instead use the cs.WAL on cs.Start, // but we'd have to allow the WAL to replay a block that wrote it's ENDHEIGHT log.Info("Replay last block using real app") - return h.replayLastBlock(storeBlockHeight, proxyApp.Consensus()) + return h.replayBlock(storeBlockHeight, proxyApp.Consensus()) } else if appBlockHeight == storeBlockHeight { - // We ran Commit, but didn't save the state, so ApplyBlock with mock app + // We ran Commit, but didn't save the state, so replayBlock with mock app abciResponses := h.state.LoadABCIResponses() mockApp := newMockProxyApp(appHash, abciResponses) log.Info("Replay last block using mock app") - return h.replayLastBlock(storeBlockHeight, mockApp) + return h.replayBlock(storeBlockHeight, mockApp) } } @@ -295,18 +295,18 @@ func (h *Handshaker) ReplayBlocks(appHash []byte, appBlockHeight int, proxyApp p return nil, nil } -func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, useReplayFunc bool) ([]byte, error) { +func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, mutateState bool) ([]byte, error) { // App is further behind than it should be, so we need to replay blocks. // We replay all blocks from appBlockHeight+1. - // If useReplayFunc == true, stop short of the last block - // so it can be replayed using the WAL in ReplayBlocks. // Note that we don't have an old version of the state, - // so we by-pass state validation using sm.ApplyBlock. + // so we by-pass state validation/mutation using sm.ApplyBlock. + // If mutateState == true, stop short of the last block + // so it can be replayed with a real state.ApplyBlock var appHash []byte var err error finalBlock := storeBlockHeight - if useReplayFunc { + if mutateState { finalBlock -= 1 } for i := appBlockHeight + 1; i <= finalBlock; i++ { @@ -320,7 +320,7 @@ func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, store h.nBlocks += 1 } - if useReplayFunc { + if mutateState { // sync the final block return h.ReplayBlocks(appHash, finalBlock, proxyApp) } @@ -329,7 +329,7 @@ func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, store } // ApplyBlock on the proxyApp with the last block. -func (h *Handshaker) replayLastBlock(height int, proxyApp proxy.AppConnConsensus) ([]byte, error) { +func (h *Handshaker) replayBlock(height int, proxyApp proxy.AppConnConsensus) ([]byte, error) { mempool := types.MockMempool{} var eventCache types.Fireable // nil diff --git a/consensus/state.go b/consensus/state.go index f1423e2f2..6ff97ddc8 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -1217,8 +1217,12 @@ func (cs *ConsensusState) finalizeCommit(height int) { fail.Fail() // XXX // Finish writing to the WAL for this height. - // NOTE: ConsensusState should not be started again - // until we successfully call ApplyBlock (eg. in Handshake after restart) + // NOTE: If we fail before writing this, we'll never write it, + // and just recover by running ApplyBlock in the Handshake. + // If we moved it before persisting the block, we'd have to allow + // WAL replay for blocks with an #ENDHEIGHT + // 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) } @@ -1244,9 +1248,10 @@ func (cs *ConsensusState) finalizeCommit(height int) { // Fire event for new block. // NOTE: If we fail before firing, these events will never fire // - // Some options (for which they may fire more than once. I guess that's fine): + // TODO: Either // * Fire before persisting state, in ApplyBlock // * Fire on start up if we haven't written any new WAL msgs + // Both options mean we may fire more than once. Is that fine ? types.FireEventNewBlock(cs.evsw, types.EventDataNewBlock{block}) types.FireEventNewBlockHeader(cs.evsw, types.EventDataNewBlockHeader{block.Header}) eventCache.Flush() @@ -1256,6 +1261,8 @@ func (cs *ConsensusState) finalizeCommit(height int) { // NewHeightStep! cs.updateToState(stateCopy) + fail.Fail() // XXX + // cs.StartTime is already set. // Schedule Round0 to start soon. cs.scheduleRound0(&cs.RoundState) diff --git a/state/execution.go b/state/execution.go index d978c1956..959231eed 100644 --- a/state/execution.go +++ b/state/execution.go @@ -88,19 +88,14 @@ func execBlockOnProxyApp(eventCache types.Fireable, proxyAppConn proxy.AppConnCo return nil, err } - fail.Fail() // XXX - // Run txs of block for _, tx := range block.Txs { - fail.FailRand(len(block.Txs)) // XXX proxyAppConn.DeliverTxAsync(tx) if err := proxyAppConn.Error(); err != nil { return nil, err } } - fail.Fail() // XXX - // End block abciResponses.EndBlock, err = proxyAppConn.EndBlockSync(uint64(block.Height)) if err != nil { @@ -108,8 +103,6 @@ func execBlockOnProxyApp(eventCache types.Fireable, proxyAppConn proxy.AppConnCo return nil, err } - fail.Fail() // XXX - valDiff := abciResponses.EndBlock.Diffs log.Info("Executed block", "height", block.Height, "valid txs", validTxs, "invalid txs", invalidTxs) @@ -292,9 +285,8 @@ func (s *State) indexTxs(abciResponses *ABCIResponses) { s.TxIndexer.Batch(batch) } -// Apply and commit a block, but without all the state validation. +// Apply and commit a block on the proxyApp without validating or mutating the state // Returns the application root hash (result of abci.Commit) -// TODO handle abciResponses func ApplyBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block) ([]byte, error) { var eventCache types.Fireable // nil _, err := execBlockOnProxyApp(eventCache, appConnConsensus, block) diff --git a/state/state.go b/state/state.go index eb7159e24..7d0f24558 100644 --- a/state/state.go +++ b/state/state.go @@ -186,7 +186,7 @@ type ABCIResponses struct { DeliverTx []*abci.ResponseDeliverTx EndBlock abci.ResponseEndBlock - txs types.Txs // for reference later + txs types.Txs // reference for indexing results by hash } func NewABCIResponses(block *types.Block) *ABCIResponses {