rollback command doesn't rollback state if block store has one more block

it's not very useful under the app hash mismatch scenario.
this patch will rollback the block store in such case.

changelog

CHANGELOG_PENDING.md

@@ -73,6 +73,7 @@ Special thanks to external contributors on this release:
 - [rpc] \#7612 paginate mempool /unconfirmed_txs rpc endpoint (@spacech1mp)
 - [light] [\#7536](https://github.com/tendermint/tendermint/pull/7536) rpc /status call returns info about the light client (@jmalicevic)
 - [types] \#7765 Replace EvidenceData with EvidenceList to avoid unnecessary nesting of evidence fields within a block. (@jmalicevic)
+- [cli] [\#8101](https://github.com/tendermint/tendermint/pull/8101) rollback command rollback block store if it has one more block than block state.
 
 ### BUG FIXES
 

internal/consensus/replay_test.go

@@ -1231,6 +1231,10 @@ func (bs *mockBlockStore) PruneBlocks(height int64) (uint64, error) {
 	return pruned, nil
 }
 
+func (bs *mockBlockStore) Rollback() error {
+	return nil
+}
+
 //---------------------------------------
 // Test handshake/init chain
 

internal/state/mocks/block_store.go

@@ -208,3 +208,8 @@ func (_m *BlockStore) Size() int64 {
 
 	return r0
 }
+
+func (_m *BlockStore) Rollback() error {
+	_m.Called()
+	return nil
+}

internal/state/rollback.go

@@ -23,9 +23,12 @@ func Rollback(bs BlockStore, ss Store) (int64, []byte, error) {
 
 	// NOTE: persistence of state and blocks don't happen atomically. Therefore it is possible that
 	// when the user stopped the node the state wasn't updated but the blockstore was. In this situation
-	// we don't need to rollback any state and can just return early
+	// we need to rollback the block store too.
 	if height == invalidState.LastBlockHeight+1 {
-		return invalidState.LastBlockHeight, invalidState.AppHash, nil
+		if err := bs.Rollback(); err != nil {
+			return -1, nil, err
+		}
+		height--
 	}
 
 	// If the state store isn't one below nor equal to the blockstore height than this violates the

internal/state/services.go

@@ -36,6 +36,8 @@ type BlockStore interface {
 
 	LoadBlockCommit(height int64) *types.Commit
 	LoadSeenCommit() *types.Commit
+
+	Rollback() error
 }
 
 //-----------------------------------------------------------------------------

internal/store/store.go

@@ -313,28 +313,6 @@ func (bs *BlockStore) PruneBlocks(height int64) (uint64, error) {
 		return 0, fmt.Errorf("height must be equal to or less than the latest height %d", bs.Height())
 	}
 
-	// when removing the block meta, use the hash to remove the hash key at the same time
-	removeBlockHash := func(key, value []byte, batch dbm.Batch) error {
-		// unmarshal block meta
-		var pbbm = new(tmproto.BlockMeta)
-		err := proto.Unmarshal(value, pbbm)
-		if err != nil {
-			return fmt.Errorf("unmarshal to tmproto.BlockMeta: %w", err)
-		}
-
-		blockMeta, err := types.BlockMetaFromProto(pbbm)
-		if err != nil {
-			return fmt.Errorf("error from proto blockMeta: %w", err)
-		}
-
-		// delete the hash key corresponding to the block meta's hash
-		if err := batch.Delete(blockHashKey(blockMeta.BlockID.Hash)); err != nil {
-			return fmt.Errorf("failed to delete hash key: %X: %w", blockHashKey(blockMeta.BlockID.Hash), err)
-		}
-
-		return nil
-	}
-
 	// remove block meta first as this is used to indicate whether the block exists.
 	// For this reason, we also use ony block meta as a measure of the amount of blocks pruned
 	pruned, err := bs.pruneRange(blockMetaKey(0), blockMetaKey(height), removeBlockHash)
@@ -353,6 +331,28 @@ func (bs *BlockStore) PruneBlocks(height int64) (uint64, error) {
 	return pruned, nil
 }
 
+// when removing the block meta, use the hash to remove the hash key at the same time
+func removeBlockHash(key, value []byte, batch dbm.Batch) error {
+	// unmarshal block meta
+	var pbbm = new(tmproto.BlockMeta)
+	err := proto.Unmarshal(value, pbbm)
+	if err != nil {
+		return fmt.Errorf("unmarshal to tmproto.BlockMeta: %w", err)
+	}
+
+	blockMeta, err := types.BlockMetaFromProto(pbbm)
+	if err != nil {
+		return fmt.Errorf("error from proto blockMeta: %w", err)
+	}
+
+	// delete the hash key corresponding to the block meta's hash
+	if err := batch.Delete(blockHashKey(blockMeta.BlockID.Hash)); err != nil {
+		return fmt.Errorf("failed to delete hash key: %X: %w", blockHashKey(blockMeta.BlockID.Hash), err)
+	}
+
+	return nil
+}
+
 // pruneRange is a generic function for deleting a range of values based on the lowest
 // height up to but excluding retainHeight. For each key/value pair, an optional hook can be
 // executed before the deletion itself is made. pruneRange will use batch delete to delete
@@ -576,6 +576,33 @@ func (bs *BlockStore) Close() error {
 	return bs.db.Close()
 }
 
+// Rollback rollbacks the latest block from BlockStore.
+func (bs *BlockStore) Rollback() error {
+	height := bs.Height()
+	if height <= 0 {
+		return fmt.Errorf("can't rollback height %v", height)
+	}
+
+	pruned, err := bs.pruneRange(blockMetaKey(height), blockMetaKey(height+1), removeBlockHash)
+	if err != nil {
+		return err
+	}
+
+	if pruned != 1 {
+		return fmt.Errorf("the number of rollbacked blocks don't match %v", pruned)
+	}
+
+	if _, err := bs.pruneRange(blockPartKey(height, 0), blockPartKey(height+1, 0), nil); err != nil {
+		return err
+	}
+
+	if _, err := bs.pruneRange(blockCommitKey(height), blockCommitKey(height+1), nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 //---------------------------------- KEY ENCODING -----------------------------------------
 
 // key prefixes