abci: Move app_hash parameter from Commit to FinalizeBlock (#8664)

* Removed from proto

* make proto-gen

* make build works

* make some tests pass

* Fix TestMempoolTxConcurrentWithCommit

* Minor change

* Update abci/types/types.go

* Update internal/state/execution.go

* Update test/e2e/app/state.go

Co-authored-by: Callum Waters <cmwaters19@gmail.com>

* Updated changelog and `UPGRADING.md`

* Fixed abci-cli tests, and doc

* Addressed @cmwaters' comments

* Addressed @cmwaters' comments, part 2

Co-authored-by: Callum Waters <cmwaters19@gmail.com>

internal/consensus/mempool_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -136,8 +137,10 @@ func checkTxsRange(ctx context.Context, t *testing.T, cs *State, start, end int)
 	for i := start; i < end; i++ {
 		txBytes := make([]byte, 8)
 		binary.BigEndian.PutUint64(txBytes, uint64(i))
-		err := assertMempool(t, cs.txNotifier).CheckTx(ctx, txBytes, nil, mempool.TxInfo{})
+		var rCode uint32
+		err := assertMempool(t, cs.txNotifier).CheckTx(ctx, txBytes, func(r *abci.ResponseCheckTx) { rCode = r.Code }, mempool.TxInfo{})
 		require.NoError(t, err, "error after checkTx")
+		require.Equal(t, code.CodeTypeOK, rCode, "checkTx code is error, txBytes %X", txBytes)
 	}
 }
 
@@ -173,6 +176,7 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
 		case msg := <-newBlockHeaderCh:
 			headerEvent := msg.Data().(types.EventDataNewBlockHeader)
 			n += headerEvent.NumTxs
+			logger.Info("new transactions", "nTxs", headerEvent.NumTxs, "total", n)
 		case <-time.After(30 * time.Second):
 			t.Fatal("Timed out waiting 30s to commit blocks with transactions")
 		}
@@ -202,10 +206,10 @@ func TestMempoolRmBadTx(t *testing.T) {
 	resFinalize, err := app.FinalizeBlock(ctx, &abci.RequestFinalizeBlock{Txs: [][]byte{txBytes}})
 	require.NoError(t, err)
 	assert.False(t, resFinalize.TxResults[0].IsErr(), fmt.Sprintf("expected no error. got %v", resFinalize))
+	assert.True(t, len(resFinalize.AppHash) > 0)
 
-	resCommit, err := app.Commit(ctx)
+	_, err = app.Commit(ctx)
 	require.NoError(t, err)
-	assert.True(t, len(resCommit.Data) > 0)
 
 	emptyMempoolCh := make(chan struct{})
 	checkTxRespCh := make(chan struct{})
@@ -263,6 +267,7 @@ type CounterApplication struct {
 
 	txCount        int
 	mempoolTxCount int
+	mu             sync.Mutex
 }
 
 func NewCounterApplication() *CounterApplication {
@@ -270,10 +275,16 @@ func NewCounterApplication() *CounterApplication {
 }
 
 func (app *CounterApplication) Info(_ context.Context, req *abci.RequestInfo) (*abci.ResponseInfo, error) {
+	app.mu.Lock()
+	defer app.mu.Unlock()
+
 	return &abci.ResponseInfo{Data: fmt.Sprintf("txs:%v", app.txCount)}, nil
 }
 
 func (app *CounterApplication) FinalizeBlock(_ context.Context, req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
+	app.mu.Lock()
+	defer app.mu.Unlock()
+
 	respTxs := make([]*abci.ExecTxResult, len(req.Txs))
 	for i, tx := range req.Txs {
 		txValue := txAsUint64(tx)
@@ -287,10 +298,21 @@ func (app *CounterApplication) FinalizeBlock(_ context.Context, req *abci.Reques
 		app.txCount++
 		respTxs[i] = &abci.ExecTxResult{Code: code.CodeTypeOK}
 	}
-	return &abci.ResponseFinalizeBlock{TxResults: respTxs}, nil
+
+	res := &abci.ResponseFinalizeBlock{TxResults: respTxs}
+
+	if app.txCount > 0 {
+		res.AppHash = make([]byte, 8)
+		binary.BigEndian.PutUint64(res.AppHash, uint64(app.txCount))
+	}
+
+	return res, nil
 }
 
 func (app *CounterApplication) CheckTx(_ context.Context, req *abci.RequestCheckTx) (*abci.ResponseCheckTx, error) {
+	app.mu.Lock()
+	defer app.mu.Unlock()
+
 	txValue := txAsUint64(req.Tx)
 	if txValue != uint64(app.mempoolTxCount) {
 		return &abci.ResponseCheckTx{
@@ -308,13 +330,11 @@ func txAsUint64(tx []byte) uint64 {
 }
 
 func (app *CounterApplication) Commit(context.Context) (*abci.ResponseCommit, error) {
+	app.mu.Lock()
+	defer app.mu.Unlock()
+
 	app.mempoolTxCount = app.txCount
-	if app.txCount == 0 {
-		return &abci.ResponseCommit{}, nil
-	}
-	hash := make([]byte, 8)
-	binary.BigEndian.PutUint64(hash, uint64(app.txCount))
-	return &abci.ResponseCommit{Data: hash}, nil
+	return &abci.ResponseCommit{}, nil
 }
 
 func (app *CounterApplication) PrepareProposal(_ context.Context, req *abci.RequestPrepareProposal) (*abci.ResponsePrepareProposal, error) {

internal/consensus/replay_stubs.go

@@ -86,5 +86,5 @@ func (mock *mockProxyApp) FinalizeBlock(_ context.Context, req *abci.RequestFina
 }
 
 func (mock *mockProxyApp) Commit(context.Context) (*abci.ResponseCommit, error) {
-	return &abci.ResponseCommit{Data: mock.appHash}, nil
+	return &abci.ResponseCommit{}, nil
 }

internal/consensus/replay_test.go

@@ -1017,15 +1017,15 @@ type badApp struct {
 	onlyLastHashIsWrong bool
 }
 
-func (app *badApp) Commit(context.Context) (*abci.ResponseCommit, error) {
+func (app *badApp) FinalizeBlock(_ context.Context, _ *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
 	app.height++
 	if app.onlyLastHashIsWrong {
 		if app.height == app.numBlocks {
-			return &abci.ResponseCommit{Data: tmrand.Bytes(8)}, nil
+			return &abci.ResponseFinalizeBlock{AppHash: tmrand.Bytes(8)}, nil
 		}
-		return &abci.ResponseCommit{Data: []byte{app.height}}, nil
+		return &abci.ResponseFinalizeBlock{AppHash: []byte{app.height}}, nil
 	} else if app.allHashesAreWrong {
-		return &abci.ResponseCommit{Data: tmrand.Bytes(8)}, nil
+		return &abci.ResponseFinalizeBlock{AppHash: tmrand.Bytes(8)}, nil
 	}
 
 	panic("either allHashesAreWrong or onlyLastHashIsWrong must be set")

internal/state/execution.go

@@ -206,7 +206,7 @@ func (blockExec *BlockExecutor) ApplyBlock(
 		return state, ErrInvalidBlock(err)
 	}
 	startTime := time.Now().UnixNano()
-	finalizeBlockResponse, err := blockExec.appClient.FinalizeBlock(
+	fBlockRes, err := blockExec.appClient.FinalizeBlock(
 		ctx,
 		&abci.RequestFinalizeBlock{
 			Hash:                block.Hash(),
@@ -225,8 +225,16 @@ func (blockExec *BlockExecutor) ApplyBlock(
 		return state, ErrProxyAppConn(err)
 	}
 
+	blockExec.logger.Info(
+		"finalized block",
+		"height", block.Height,
+		"num_txs_res", len(fBlockRes.TxResults),
+		"num_val_updates", len(fBlockRes.ValidatorUpdates),
+		"block_app_hash", fmt.Sprintf("%X", fBlockRes.AppHash),
+	)
+
 	abciResponses := &tmstate.ABCIResponses{
-		FinalizeBlock: finalizeBlockResponse,
+		FinalizeBlock: fBlockRes,
 	}
 
 	// Save the results before we commit.
@@ -235,12 +243,12 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	}
 
 	// validate the validator updates and convert to tendermint types
-	err = validateValidatorUpdates(finalizeBlockResponse.ValidatorUpdates, state.ConsensusParams.Validator)
+	err = validateValidatorUpdates(fBlockRes.ValidatorUpdates, state.ConsensusParams.Validator)
 	if err != nil {
 		return state, fmt.Errorf("error in validator updates: %w", err)
 	}
 
-	validatorUpdates, err := types.PB2TM.ValidatorUpdates(finalizeBlockResponse.ValidatorUpdates)
+	validatorUpdates, err := types.PB2TM.ValidatorUpdates(fBlockRes.ValidatorUpdates)
 	if err != nil {
 		return state, err
 	}
@@ -248,23 +256,23 @@ func (blockExec *BlockExecutor) ApplyBlock(
 		blockExec.logger.Debug("updates to validators", "updates", types.ValidatorListString(validatorUpdates))
 		blockExec.metrics.ValidatorSetUpdates.Add(1)
 	}
-	if finalizeBlockResponse.ConsensusParamUpdates != nil {
+	if fBlockRes.ConsensusParamUpdates != nil {
 		blockExec.metrics.ConsensusParamUpdates.Add(1)
 	}
 
 	// Update the state with the block and responses.
-	rs, err := abci.MarshalTxResults(finalizeBlockResponse.TxResults)
+	rs, err := abci.MarshalTxResults(fBlockRes.TxResults)
 	if err != nil {
 		return state, fmt.Errorf("marshaling TxResults: %w", err)
 	}
 	h := merkle.HashFromByteSlices(rs)
-	state, err = state.Update(blockID, &block.Header, h, finalizeBlockResponse.ConsensusParamUpdates, validatorUpdates)
+	state, err = state.Update(blockID, &block.Header, h, fBlockRes.ConsensusParamUpdates, validatorUpdates)
 	if err != nil {
 		return state, fmt.Errorf("commit failed for application: %w", err)
 	}
 
 	// Lock mempool, commit app state, update mempoool.
-	appHash, retainHeight, err := blockExec.Commit(ctx, state, block, finalizeBlockResponse.TxResults)
+	retainHeight, err := blockExec.Commit(ctx, state, block, fBlockRes.TxResults)
 	if err != nil {
 		return state, fmt.Errorf("commit failed for application: %w", err)
 	}
@@ -273,7 +281,7 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	blockExec.evpool.Update(ctx, state, block.Evidence)
 
 	// Update the app hash and save the state.
-	state.AppHash = appHash
+	state.AppHash = fBlockRes.AppHash
 	if err := blockExec.store.Save(state); err != nil {
 		return state, err
 	}
@@ -293,7 +301,7 @@ func (blockExec *BlockExecutor) ApplyBlock(
 
 	// Events are fired after everything else.
 	// NOTE: if we crash between Commit and Save, events wont be fired during replay
-	fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, finalizeBlockResponse, validatorUpdates)
+	fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, fBlockRes, validatorUpdates)
 
 	return state, nil
 }
@@ -338,7 +346,7 @@ func (blockExec *BlockExecutor) Commit(
 	state State,
 	block *types.Block,
 	txResults []*abci.ExecTxResult,
-) ([]byte, int64, error) {
+) (int64, error) {
 	blockExec.mempool.Lock()
 	defer blockExec.mempool.Unlock()
 
@@ -347,14 +355,14 @@ func (blockExec *BlockExecutor) Commit(
 	err := blockExec.mempool.FlushAppConn(ctx)
 	if err != nil {
 		blockExec.logger.Error("client error during mempool.FlushAppConn", "err", err)
-		return nil, 0, err
+		return 0, err
 	}
 
 	// Commit block, get hash back
 	res, err := blockExec.appClient.Commit(ctx)
 	if err != nil {
 		blockExec.logger.Error("client error during proxyAppConn.Commit", "err", err)
-		return nil, 0, err
+		return 0, err
 	}
 
 	// ResponseCommit has no error code - just data
@@ -362,7 +370,7 @@ func (blockExec *BlockExecutor) Commit(
 		"committed state",
 		"height", block.Height,
 		"num_txs", len(block.Txs),
-		"app_hash", fmt.Sprintf("%X", res.Data),
+		"block_app_hash", fmt.Sprintf("%X", block.AppHash),
 	)
 
 	// Update mempool.
@@ -376,7 +384,7 @@ func (blockExec *BlockExecutor) Commit(
 		state.ConsensusParams.ABCI.RecheckTx,
 	)
 
-	return res.Data, res.RetainHeight, err
+	return res.RetainHeight, err
 }
 
 func buildLastCommitInfo(block *types.Block, store Store, initialHeight int64) abci.CommitInfo {
@@ -708,15 +716,15 @@ func ExecCommitBlock(
 		fireEvents(be.logger, be.eventBus, block, blockID, finalizeBlockResponse, validatorUpdates)
 	}
 
-	// Commit block, get hash back
-	res, err := appConn.Commit(ctx)
+	// Commit block
+	_, err = appConn.Commit(ctx)
 	if err != nil {
-		logger.Error("client error during proxyAppConn.Commit", "err", res)
+		logger.Error("client error during proxyAppConn.Commit", "err", err)
 		return nil, err
 	}
 
-	// ResponseCommit has no error or log, just data
-	return res.Data, nil
+	// ResponseCommit has no error or log
+	return finalizeBlockResponse.AppHash, nil
 }
 
 func (blockExec *BlockExecutor) pruneBlocks(retainHeight int64) (uint64, error) {