add support for block pruning via ABCI Commit response (#4588)

* Added BlockStore.DeleteBlock()

* Added initial block pruner prototype

* wip

* Added BlockStore.PruneBlocks()

* Added consensus setting for block pruning

* Added BlockStore base

* Error on replay if base does not have blocks

* Handle missing blocks when sending VoteSetMaj23Message

* Error message tweak

* Properly update blockstore state

* Error message fix again

* blockchain: ignore peer missing blocks

* Added FIXME

* Added test for block replay with truncated history

* Handle peer base in blockchain reactor

* Improved replay error handling

* Added tests for Store.PruneBlocks()

* Fix non-RPC handling of truncated block history

* Panic on missing block meta in needProofBlock()

* Updated changelog

* Handle truncated block history in RPC layer

* Added info about earliest block in /status RPC

* Reorder height and base in blockchain reactor messages

* Updated changelog

* Fix tests

* Appease linter

* Minor review fixes

* Non-empty BlockStores should always have base > 0

* Update code to assume base > 0 invariant

* Added blockstore tests for pruning to 0

* Make sure we don't prune below the current base

* Added BlockStore.Size()

* config: added retain_blocks recommendations

* Update v1 blockchain reactor to handle blockstore base

* Added state database pruning

* Propagate errors on missing validator sets

* Comment tweaks

* Improved error message

Co-Authored-By: Anton Kaliaev <anton.kalyaev@gmail.com>

* use ABCI field ResponseCommit.retain_height instead of retain-blocks config option

* remove State.RetainHeight, return value instead

* fix minor issues

* rename pruneHeights() to pruneBlocks()

* noop to fix GitHub borkage

Co-authored-by: Anton Kaliaev <anton.kalyaev@gmail.com>

state/errors.go

@@ -21,6 +21,11 @@ type (
 		AppHeight  int64
 	}
 
+	ErrAppBlockHeightTooLow struct {
+		AppHeight int64
+		StoreBase int64
+	}
+
 	ErrLastStateMismatch struct {
 		Height int64
 		Core   []byte
@@ -46,12 +51,12 @@ type (
 )
 
 func (e ErrUnknownBlock) Error() string {
-	return fmt.Sprintf("Could not find block #%d", e.Height)
+	return fmt.Sprintf("could not find block #%d", e.Height)
 }
 
 func (e ErrBlockHashMismatch) Error() string {
 	return fmt.Sprintf(
-		"App block hash (%X) does not match core block hash (%X) for height %d",
+		"app block hash (%X) does not match core block hash (%X) for height %d",
 		e.AppHash,
 		e.CoreHash,
 		e.Height,
@@ -59,11 +64,16 @@ func (e ErrBlockHashMismatch) Error() string {
 }
 
 func (e ErrAppBlockHeightTooHigh) Error() string {
-	return fmt.Sprintf("App block height (%d) is higher than core (%d)", e.AppHeight, e.CoreHeight)
+	return fmt.Sprintf("app block height (%d) is higher than core (%d)", e.AppHeight, e.CoreHeight)
 }
+
+func (e ErrAppBlockHeightTooLow) Error() string {
+	return fmt.Sprintf("app block height (%d) is too far below block store base (%d)", e.AppHeight, e.StoreBase)
+}
+
 func (e ErrLastStateMismatch) Error() string {
 	return fmt.Sprintf(
-		"Latest tendermint block (%d) LastAppHash (%X) does not match app's AppHash (%X)",
+		"latest tendermint block (%d) LastAppHash (%X) does not match app's AppHash (%X)",
 		e.Height,
 		e.Core,
 		e.App,
@@ -72,20 +82,20 @@ func (e ErrLastStateMismatch) Error() string {
 
 func (e ErrStateMismatch) Error() string {
 	return fmt.Sprintf(
-		"State after replay does not match saved state. Got ----\n%v\nExpected ----\n%v\n",
+		"state after replay does not match saved state. Got ----\n%v\nExpected ----\n%v\n",
 		e.Got,
 		e.Expected,
 	)
 }
 
 func (e ErrNoValSetForHeight) Error() string {
-	return fmt.Sprintf("Could not find validator set for height #%d", e.Height)
+	return fmt.Sprintf("could not find validator set for height #%d", e.Height)
 }
 
 func (e ErrNoConsensusParamsForHeight) Error() string {
-	return fmt.Sprintf("Could not find consensus params for height #%d", e.Height)
+	return fmt.Sprintf("could not find consensus params for height #%d", e.Height)
 }
 
 func (e ErrNoABCIResponsesForHeight) Error() string {
-	return fmt.Sprintf("Could not find results for height #%d", e.Height)
+	return fmt.Sprintf("could not find results for height #%d", e.Height)
 }

state/execution.go

@@ -119,13 +119,14 @@ func (blockExec *BlockExecutor) ValidateBlock(state State, block *types.Block) e
 
 // ApplyBlock validates the block against the state, executes it against the app,
 // fires the relevant events, commits the app, and saves the new state and responses.
+// It returns the new state and the block height to retain (pruning older blocks).
 // It's the only function that needs to be called
 // from outside this package to process and commit an entire block.
 // It takes a blockID to avoid recomputing the parts hash.
-func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, block *types.Block) (State, error) {
+func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, block *types.Block) (State, int64, error) {
 
 	if err := blockExec.ValidateBlock(state, block); err != nil {
-		return state, ErrInvalidBlock(err)
+		return state, 0, ErrInvalidBlock(err)
 	}
 
 	startTime := time.Now().UnixNano()
@@ -133,7 +134,7 @@ func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, b
 	endTime := time.Now().UnixNano()
 	blockExec.metrics.BlockProcessingTime.Observe(float64(endTime-startTime) / 1000000)
 	if err != nil {
-		return state, ErrProxyAppConn(err)
+		return state, 0, ErrProxyAppConn(err)
 	}
 
 	fail.Fail() // XXX
@@ -147,11 +148,11 @@ func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, b
 	abciValUpdates := abciResponses.EndBlock.ValidatorUpdates
 	err = validateValidatorUpdates(abciValUpdates, state.ConsensusParams.Validator)
 	if err != nil {
-		return state, fmt.Errorf("error in validator updates: %v", err)
+		return state, 0, fmt.Errorf("error in validator updates: %v", err)
 	}
 	validatorUpdates, err := types.PB2TM.ValidatorUpdates(abciValUpdates)
 	if err != nil {
-		return state, err
+		return state, 0, err
 	}
 	if len(validatorUpdates) > 0 {
 		blockExec.logger.Info("Updates to validators", "updates", types.ValidatorListString(validatorUpdates))
@@ -160,13 +161,13 @@ func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, b
 	// Update the state with the block and responses.
 	state, err = updateState(state, blockID, &block.Header, abciResponses, validatorUpdates)
 	if err != nil {
-		return state, fmt.Errorf("commit failed for application: %v", err)
+		return state, 0, fmt.Errorf("commit failed for application: %v", err)
 	}
 
 	// Lock mempool, commit app state, update mempoool.
-	appHash, err := blockExec.Commit(state, block, abciResponses.DeliverTxs)
+	appHash, retainHeight, err := blockExec.Commit(state, block, abciResponses.DeliverTxs)
 	if err != nil {
-		return state, fmt.Errorf("commit failed for application: %v", err)
+		return state, 0, fmt.Errorf("commit failed for application: %v", err)
 	}
 
 	// Update evpool with the block and state.
@@ -184,12 +185,12 @@ func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, b
 	// NOTE: if we crash between Commit and Save, events wont be fired during replay
 	fireEvents(blockExec.logger, blockExec.eventBus, block, abciResponses, validatorUpdates)
 
-	return state, nil
+	return state, retainHeight, nil
 }
 
 // Commit locks the mempool, runs the ABCI Commit message, and updates the
 // mempool.
-// It returns the result of calling abci.Commit (the AppHash), and an error.
+// It returns the result of calling abci.Commit (the AppHash) and the height to retain (if any).
 // The Mempool must be locked during commit and update because state is
 // typically reset on Commit and old txs must be replayed against committed
 // state before new txs are run in the mempool, lest they be invalid.
@@ -197,7 +198,7 @@ func (blockExec *BlockExecutor) Commit(
 	state State,
 	block *types.Block,
 	deliverTxResponses []*abci.ResponseDeliverTx,
-) ([]byte, error) {
+) ([]byte, int64, error) {
 	blockExec.mempool.Lock()
 	defer blockExec.mempool.Unlock()
 
@@ -206,7 +207,7 @@ func (blockExec *BlockExecutor) Commit(
 	err := blockExec.mempool.FlushAppConn()
 	if err != nil {
 		blockExec.logger.Error("Client error during mempool.FlushAppConn", "err", err)
-		return nil, err
+		return nil, 0, err
 	}
 
 	// Commit block, get hash back
@@ -216,7 +217,7 @@ func (blockExec *BlockExecutor) Commit(
 			"Client error during proxyAppConn.CommitSync",
 			"err", err,
 		)
-		return nil, err
+		return nil, 0, err
 	}
 	// ResponseCommit has no error code - just data
 
@@ -236,7 +237,7 @@ func (blockExec *BlockExecutor) Commit(
 		TxPostCheck(state),
 	)
 
-	return res.Data, err
+	return res.Data, res.RetainHeight, err
 }
 
 //---------------------------------------------------------

state/execution_test.go

@@ -27,7 +27,9 @@ var (
 )
 
 func TestApplyBlock(t *testing.T) {
-	cc := proxy.NewLocalClientCreator(kvstore.NewApplication())
+	app := kvstore.NewApplication()
+	app.RetainBlocks = 1
+	cc := proxy.NewLocalClientCreator(app)
 	proxyApp := proxy.NewAppConns(cc)
 	err := proxyApp.Start()
 	require.Nil(t, err)
@@ -41,9 +43,9 @@ func TestApplyBlock(t *testing.T) {
 	block := makeBlock(state, 1)
 	blockID := types.BlockID{Hash: block.Hash(), PartsHeader: block.MakePartSet(testPartSize).Header()}
 
-	//nolint:ineffassign
-	state, err = blockExec.ApplyBlock(state, blockID, block)
+	_, retainHeight, err := blockExec.ApplyBlock(state, blockID, block)
 	require.Nil(t, err)
+	assert.EqualValues(t, retainHeight, 1)
 
 	// TODO check state and mempool
 }
@@ -356,7 +358,7 @@ func TestEndBlockValidatorUpdates(t *testing.T) {
 		{PubKey: types.TM2PB.PubKey(pubkey), Power: 10},
 	}
 
-	state, err = blockExec.ApplyBlock(state, blockID, block)
+	state, _, err = blockExec.ApplyBlock(state, blockID, block)
 	require.Nil(t, err)
 
 	// test new validator was added to NextValidators
@@ -410,7 +412,7 @@ func TestEndBlockValidatorUpdatesResultingInEmptySet(t *testing.T) {
 		{PubKey: types.TM2PB.PubKey(state.Validators.Validators[0].PubKey), Power: 0},
 	}
 
-	assert.NotPanics(t, func() { state, err = blockExec.ApplyBlock(state, blockID, block) })
+	assert.NotPanics(t, func() { state, _, err = blockExec.ApplyBlock(state, blockID, block) })
 	assert.NotNil(t, err)
 	assert.NotEmpty(t, state.NextValidators.Validators)
 

state/helpers_test.go

@@ -66,7 +66,7 @@ func makeAndApplyGoodBlock(state sm.State, height int64, lastCommit *types.Commi
 	}
 	blockID := types.BlockID{Hash: block.Hash(),
 		PartsHeader: types.PartSetHeader{Total: 3, Hash: tmrand.Bytes(32)}}
-	state, err := blockExec.ApplyBlock(state, blockID, block)
+	state, _, err := blockExec.ApplyBlock(state, blockID, block)
 	if err != nil {
 		return state, types.BlockID{}, err
 	}

state/services.go

@@ -14,13 +14,17 @@ import (
 
 // BlockStore defines the interface used by the ConsensusState.
 type BlockStore interface {
+	Base() int64
 	Height() int64
+	Size() int64
 
 	LoadBlockMeta(height int64) *types.BlockMeta
 	LoadBlock(height int64) *types.Block
 
 	SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit)
 
+	PruneBlocks(height int64) (uint64, error)
+
 	LoadBlockByHash(hash []byte) *types.Block
 	LoadBlockPart(height int64, index int) *types.Part
 

state/store.go

@@ -125,6 +125,102 @@ type ABCIResponses struct {
 	BeginBlock *abci.ResponseBeginBlock  `json:"begin_block"`
 }
 
+// PruneStates deletes states between the given heights (including from, excluding to). It is not
+// guaranteed to delete all states, since the last checkpointed state and states being pointed to by
+// e.g. `LastHeightChanged` must remain. The state at to must also exist.
+//
+// The from parameter is necessary since we can't do a key scan in a performant way due to the key
+// encoding not preserving ordering: https://github.com/tendermint/tendermint/issues/4567
+// This will cause some old states to be left behind when doing incremental partial prunes,
+// specifically older checkpoints and LastHeightChanged targets.
+func PruneStates(db dbm.DB, from int64, to int64) error {
+	if from <= 0 || to <= 0 {
+		return fmt.Errorf("from height %v and to height %v must be greater than 0", from, to)
+	}
+	if from >= to {
+		return fmt.Errorf("from height %v must be lower than to height %v", from, to)
+	}
+	valInfo := loadValidatorsInfo(db, to)
+	if valInfo == nil {
+		return fmt.Errorf("validators at height %v not found", to)
+	}
+	paramsInfo := loadConsensusParamsInfo(db, to)
+	if paramsInfo == nil {
+		return fmt.Errorf("consensus params at height %v not found", to)
+	}
+
+	keepVals := make(map[int64]bool)
+	if valInfo.ValidatorSet == nil {
+		keepVals[valInfo.LastHeightChanged] = true
+		keepVals[lastStoredHeightFor(to, valInfo.LastHeightChanged)] = true // keep last checkpoint too
+	}
+	keepParams := make(map[int64]bool)
+	if paramsInfo.ConsensusParams.Equals(&types.ConsensusParams{}) {
+		keepParams[paramsInfo.LastHeightChanged] = true
+	}
+
+	batch := db.NewBatch()
+	defer batch.Close()
+	pruned := uint64(0)
+	var err error
+
+	// We have to delete in reverse order, to avoid deleting previous heights that have validator
+	// sets and consensus params that we may need to retrieve.
+	for h := to - 1; h >= from; h-- {
+		// For heights we keep, we must make sure they have the full validator set or consensus
+		// params, otherwise they will panic if they're retrieved directly (instead of
+		// indirectly via a LastHeightChanged pointer).
+		if keepVals[h] {
+			v := loadValidatorsInfo(db, h)
+			if v.ValidatorSet == nil {
+				v.ValidatorSet, err = LoadValidators(db, h)
+				if err != nil {
+					return err
+				}
+				v.LastHeightChanged = h
+				batch.Set(calcValidatorsKey(h), v.Bytes())
+			}
+		} else {
+			batch.Delete(calcValidatorsKey(h))
+		}
+
+		if keepParams[h] {
+			p := loadConsensusParamsInfo(db, h)
+			if p.ConsensusParams.Equals(&types.ConsensusParams{}) {
+				p.ConsensusParams, err = LoadConsensusParams(db, h)
+				if err != nil {
+					return err
+				}
+				p.LastHeightChanged = h
+				batch.Set(calcConsensusParamsKey(h), p.Bytes())
+			}
+		} else {
+			batch.Delete(calcConsensusParamsKey(h))
+		}
+
+		batch.Delete(calcABCIResponsesKey(h))
+		pruned++
+
+		// avoid batches growing too large by flushing to database regularly
+		if pruned%1000 == 0 && pruned > 0 {
+			err := batch.Write()
+			if err != nil {
+				return err
+			}
+			batch.Close()
+			batch = db.NewBatch()
+			defer batch.Close()
+		}
+	}
+
+	err = batch.WriteSync()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // NewABCIResponses returns a new ABCIResponses
 func NewABCIResponses(block *types.Block) *ABCIResponses {
 	resDeliverTxs := make([]*abci.ResponseDeliverTx, len(block.Data.Txs))

state/store_test.go

@@ -65,3 +65,118 @@ func BenchmarkLoadValidators(b *testing.B) {
 		})
 	}
 }
+
+func TestPruneStates(t *testing.T) {
+	testcases := map[string]struct {
+		makeHeights  int64
+		pruneFrom    int64
+		pruneTo      int64
+		expectErr    bool
+		expectVals   []int64
+		expectParams []int64
+		expectABCI   []int64
+	}{
+		"error on pruning from 0":      {100, 0, 5, true, nil, nil, nil},
+		"error when from > to":         {100, 3, 2, true, nil, nil, nil},
+		"error when from == to":        {100, 3, 3, true, nil, nil, nil},
+		"error when to does not exist": {100, 1, 101, true, nil, nil, nil},
+		"prune all":                    {100, 1, 100, false, []int64{93, 100}, []int64{95, 100}, []int64{100}},
+		"prune some":                   {10, 2, 8, false, []int64{1, 3, 8, 9, 10}, []int64{1, 5, 8, 9, 10}, []int64{1, 8, 9, 10}},
+		"prune across checkpoint":      {100001, 1, 100001, false, []int64{99993, 100000, 100001}, []int64{99995, 100001}, []int64{100001}},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			db := dbm.NewMemDB()
+
+			// Generate a bunch of state data. Validators change for heights ending with 3, and
+			// parameters when ending with 5.
+			validator := &types.Validator{Address: []byte{1, 2, 3}, VotingPower: 100}
+			validatorSet := &types.ValidatorSet{
+				Validators: []*types.Validator{validator},
+				Proposer:   validator,
+			}
+			valsChanged := int64(0)
+			paramsChanged := int64(0)
+
+			for h := int64(1); h <= tc.makeHeights; h++ {
+				if valsChanged == 0 || h%10 == 2 {
+					valsChanged = h + 1 // Have to add 1, since NextValidators is what's stored
+				}
+				if paramsChanged == 0 || h%10 == 5 {
+					paramsChanged = h
+				}
+
+				sm.SaveState(db, sm.State{
+					LastBlockHeight: h - 1,
+					Validators:      validatorSet,
+					NextValidators:  validatorSet,
+					ConsensusParams: types.ConsensusParams{
+						Block: types.BlockParams{MaxBytes: 10e6},
+					},
+					LastHeightValidatorsChanged:      valsChanged,
+					LastHeightConsensusParamsChanged: paramsChanged,
+				})
+				sm.SaveABCIResponses(db, h, sm.NewABCIResponses(&types.Block{
+					Header: types.Header{Height: h},
+					Data: types.Data{
+						Txs: types.Txs{
+							[]byte{1},
+							[]byte{2},
+							[]byte{3},
+						},
+					},
+				}))
+			}
+
+			// Test assertions
+			err := sm.PruneStates(db, tc.pruneFrom, tc.pruneTo)
+			if tc.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+
+			expectVals := sliceToMap(tc.expectVals)
+			expectParams := sliceToMap(tc.expectParams)
+			expectABCI := sliceToMap(tc.expectABCI)
+
+			for h := int64(1); h <= tc.makeHeights; h++ {
+				vals, err := sm.LoadValidators(db, h)
+				if expectVals[h] {
+					require.NoError(t, err, "validators height %v", h)
+					require.NotNil(t, vals)
+				} else {
+					require.Error(t, err, "validators height %v", h)
+					require.Equal(t, sm.ErrNoValSetForHeight{Height: h}, err)
+				}
+
+				params, err := sm.LoadConsensusParams(db, h)
+				if expectParams[h] {
+					require.NoError(t, err, "params height %v", h)
+					require.False(t, params.Equals(&types.ConsensusParams{}))
+				} else {
+					require.Error(t, err, "params height %v", h)
+					require.Equal(t, sm.ErrNoConsensusParamsForHeight{Height: h}, err)
+				}
+
+				abci, err := sm.LoadABCIResponses(db, h)
+				if expectABCI[h] {
+					require.NoError(t, err, "abci height %v", h)
+					require.NotNil(t, abci)
+				} else {
+					require.Error(t, err, "abci height %v", h)
+					require.Equal(t, sm.ErrNoABCIResponsesForHeight{Height: h}, err)
+				}
+			}
+		})
+	}
+}
+
+func sliceToMap(s []int64) map[int64]bool {
+	m := make(map[int64]bool, len(s))
+	for _, i := range s {
+		m[i] = true
+	}
+	return m
+}