Merge branch 'release/v0.50.x' into release/v0.50.x_recheck

Signed-off-by: yihuang <[email protected]>

CHANGELOG.md

@@ -54,6 +54,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 * [#261](https://github.com/crypto-org-chain/cosmos-sdk/pull/261) `ctx.BlockHeader` don't do protobuf deep copy, shallow copy seems enough, reduce scope of mutex in `PriorityNonceMempool.Remove`.
 * [#507](https://github.com/crypto-org-chain/cosmos-sdk/pull/507) mempool respect gas wanted returned by ante handler
 * [#744](https://github.com/crypto-org-chain/cosmos-sdk/pull/744) Pass raw transactions to tx executor so it can do pre-estimations.
+* [#884](https://github.com/crypto-org-chain/cosmos-sdk/pull/884) Avoid decoding tx for in PrepareProposal if it's NopMempool.
 
 ## [Unreleased]
 
@@ -65,6 +66,8 @@ Ref: https://keepachangelog.com/en/1.0.0/
 ### Improvements
 
 * (server) [#21941](https://github.com/cosmos/cosmos-sdk/pull/21941) Regenerate addrbook.json for in place testnet.
+* (store) [#923](https://github.com/crypto-org-chain/cosmos-sdk/pull/923) Enable iavl async pruning.
+* (store) [#934](https://github.com/crypto-org-chain/cosmos-sdk/pull/934) Add pause pruning.
 * (baseapp) [#20208](https://github.com/cosmos/cosmos-sdk/pull/20208) Skip running validateBasic for rechecking txs.
 
 ### Bug Fixes

baseapp/abci_utils.go

@@ -212,6 +212,9 @@ type (
 		txVerifier       ProposalTxVerifier
 		txSelector       TxSelector
 		signerExtAdapter mempool.SignerExtractionAdapter
+
+		// fastPrepareProposal together with NoOpMempool will bypass tx selector
+		fastPrepareProposal bool
 	}
 )
 
@@ -224,6 +227,12 @@ func NewDefaultProposalHandler(mp mempool.Mempool, txVerifier ProposalTxVerifier
 	}
 }
 
+func NewDefaultProposalHandlerFast(mp mempool.Mempool, txVerifier ProposalTxVerifier) *DefaultProposalHandler {
+	h := NewDefaultProposalHandler(mp, txVerifier)
+	h.fastPrepareProposal = true
+	return h
+}
+
 // SetTxSelector sets the TxSelector function on the DefaultProposalHandler.
 func (h *DefaultProposalHandler) SetTxSelector(ts TxSelector) {
 	h.txSelector = ts
@@ -264,6 +273,11 @@ func (h *DefaultProposalHandler) PrepareProposalHandler() sdk.PrepareProposalHan
 		// Note, we still need to ensure the transactions returned respect req.MaxTxBytes.
 		_, isNoOp := h.mempool.(mempool.NoOpMempool)
 		if h.mempool == nil || isNoOp {
+			if h.fastPrepareProposal {
+				txs := h.txSelector.SelectTxForProposalFast(ctx, req.Txs)
+				return &abci.ResponsePrepareProposal{Txs: txs}, nil
+			}
+
 			for _, txBz := range req.Txs {
 				tx, err := h.txVerifier.TxDecode(txBz)
 				if err != nil {
@@ -464,6 +478,12 @@ type TxSelector interface {
 	// return <true> if the caller should halt the transaction selection loop
 	// (typically over a mempool) or <false> otherwise.
 	SelectTxForProposal(ctx context.Context, maxTxBytes, maxBlockGas uint64, memTx sdk.Tx, txBz []byte, gasWanted uint64) bool
+
+	// SelectTxForProposalFast is called in the case of NoOpMempool,
+	// where cometbft already checked the block gas/size limit,
+	// so the tx selector should simply accept them all to the proposal.
+	// But extra validations on the tx are still possible though.
+	SelectTxForProposalFast(ctx context.Context, txs [][]byte) [][]byte
 }
 
 type defaultTxSelector struct {
@@ -485,7 +505,7 @@ func (ts *defaultTxSelector) SelectedTxs(_ context.Context) [][]byte {
 func (ts *defaultTxSelector) Clear() {
 	ts.totalTxBytes = 0
 	ts.totalTxGas = 0
-	ts.selectedTxs = nil
+	ts.selectedTxs = ts.selectedTxs[:0] // keep the allocated memory
 }
 
 func (ts *defaultTxSelector) SelectTxForProposal(_ context.Context, maxTxBytes, maxBlockGas uint64, memTx sdk.Tx, txBz []byte, gasWanted uint64) bool {
@@ -510,3 +530,7 @@ func (ts *defaultTxSelector) SelectTxForProposal(_ context.Context, maxTxBytes,
 	// check if we've reached capacity; if so, we cannot select any more transactions
 	return ts.totalTxBytes >= maxTxBytes || (maxBlockGas > 0 && (ts.totalTxGas >= maxBlockGas))
 }
+
+func (ts *defaultTxSelector) SelectTxForProposalFast(ctx context.Context, txs [][]byte) [][]byte {
+	return txs
+}

store/iavl/store.go

@@ -50,7 +50,7 @@ func LoadStore(db dbm.DB, logger log.Logger, key types.StoreKey, id types.Commit
 // provided DB. An error is returned if the version fails to load, or if called with a positive
 // version on an empty tree.
 func LoadStoreWithInitialVersion(db dbm.DB, logger log.Logger, key types.StoreKey, id types.CommitID, initialVersion uint64, cacheSize int, disableFastNode bool, metrics metrics.StoreMetrics) (types.CommitKVStore, error) {
-	tree := iavl.NewMutableTree(wrapper.NewDBWrapper(db), cacheSize, disableFastNode, logger, iavl.InitialVersionOption(initialVersion))
+	tree := iavl.NewMutableTree(wrapper.NewDBWrapper(db), cacheSize, disableFastNode, logger, iavl.InitialVersionOption(initialVersion), iavl.AsyncPruningOption(true))
 
 	isUpgradeable, err := tree.IsUpgradeable()
 	if err != nil {
@@ -145,6 +145,15 @@ func (st *Store) LastCommitID() types.CommitID {
 	}
 }
 
+// PausePruning implements CommitKVStore interface.
+func (st *Store) PausePruning(pause bool) {
+	if pause {
+		st.tree.SetCommitting()
+	} else {
+		st.tree.UnsetCommitting()
+	}
+}
+
 // SetPruning panics as pruning options should be provided at initialization
 // since IAVl accepts pruning options directly.
 func (st *Store) SetPruning(_ pruningtypes.PruningOptions) {

store/iavl/tree.go

@@ -22,6 +22,8 @@ type (
 		Get(key []byte) ([]byte, error)
 		Set(key, value []byte) (bool, error)
 		Remove(key []byte) ([]byte, bool, error)
+		SetCommitting()
+		UnsetCommitting()
 		SaveVersion() ([]byte, int64, error)
 		Version() int64
 		Hash() []byte
@@ -53,6 +55,14 @@ func (it *immutableTree) Remove(_ []byte) ([]byte, bool, error) {
 	panic("cannot call 'Remove' on an immutable IAVL tree")
 }
 
+func (it *immutableTree) SetCommitting() {
+	panic("cannot call 'SetCommitting' on an immutable IAVL tree")
+}
+
+func (it *immutableTree) UnsetCommitting() {
+	panic("cannot call 'UnsetCommitting' on an immutable IAVL tree")
+}
+
 func (it *immutableTree) SaveVersion() ([]byte, int64, error) {
 	panic("cannot call 'SaveVersion' on an immutable IAVL tree")
 }

store/rootmulti/store.go

@@ -464,6 +464,16 @@ func (rs *Store) LastCommitID() types.CommitID {
 	return rs.lastCommitInfo.CommitID()
 }
 
+// PausePruning temporarily pauses the pruning of all individual stores which implement
+// the PausablePruner interface.
+func (rs *Store) PausePruning(pause bool) {
+	for _, store := range rs.stores {
+		if pauseable, ok := store.(types.PausablePruner); ok {
+			pauseable.PausePruning(pause)
+		}
+	}
+}
+
 // Commit implements Committer/CommitStore.
 func (rs *Store) Commit() types.CommitID {
 	var previousHeight, version int64
@@ -485,7 +495,14 @@ func (rs *Store) Commit() types.CommitID {
 		rs.logger.Debug("commit header and version mismatch", "header_height", rs.commitHeader.Height, "version", version)
 	}
 
-	rs.lastCommitInfo = commitStores(version, rs.stores, rs.removalMap)
+	func() { // ensure unpause
+		// set the committing flag on all stores to block the pruning
+		rs.PausePruning(true)
+		// unset the committing flag on all stores to continue the pruning
+		defer rs.PausePruning(false)
+		rs.lastCommitInfo = commitStores(version, rs.stores, rs.removalMap)
+	}()
+
 	rs.lastCommitInfo.Timestamp = rs.commitHeader.Time
 	defer rs.flushMetadata(rs.db, version, rs.lastCommitInfo)
 

store/rootmulti/store_test.go

@@ -532,14 +532,18 @@ func TestMultiStore_Pruning(t *testing.T) {
 				ms.Commit()
 			}
 
-			for _, v := range tc.saved {
-				_, err := ms.CacheMultiStoreWithVersion(v)
-				require.NoError(t, err, "expected no error when loading height: %d", v)
+			for _, v := range tc.deleted {
+				// Ensure async pruning is done
+				checkErr := func() bool {
+					_, err := ms.CacheMultiStoreWithVersion(v)
+					return err != nil
+				}
+				require.Eventually(t, checkErr, 1*time.Second, 10*time.Millisecond, "expected error when loading height: %d", v)
 			}
 
-			for _, v := range tc.deleted {
+			for _, v := range tc.saved {
 				_, err := ms.CacheMultiStoreWithVersion(v)
-				require.Error(t, err, "expected error when loading height: %d", v)
+				require.NoError(t, err, "expected no error when loading height: %d", v)
 			}
 		})
 	}
@@ -564,16 +568,6 @@ func TestMultiStore_Pruning_SameHeightsTwice(t *testing.T) {
 
 	require.Equal(t, numVersions, lastCommitInfo.Version)
 
-	for v := int64(1); v < numVersions-int64(keepRecent); v++ {
-		err := ms.LoadVersion(v)
-		require.Error(t, err, "expected error when loading pruned height: %d", v)
-	}
-
-	for v := (numVersions - int64(keepRecent)); v < numVersions; v++ {
-		err := ms.LoadVersion(v)
-		require.NoError(t, err, "expected no error when loading height: %d", v)
-	}
-
 	// Get latest
 	err := ms.LoadVersion(numVersions - 1)
 	require.NoError(t, err)
@@ -588,6 +582,17 @@ func TestMultiStore_Pruning_SameHeightsTwice(t *testing.T) {
 	// Ensure that can commit one more height with no panic
 	lastCommitInfo = ms.Commit()
 	require.Equal(t, numVersions+1, lastCommitInfo.Version)
+	isPruned := func() bool {
+		ls := ms.Commit() // to flush the batch with the pruned heights
+		for v := int64(1); v < numVersions-int64(keepRecent); v++ {
+			if err := ms.LoadVersion(v); err == nil {
+				require.NoError(t, ms.LoadVersion(ls.Version)) // load latest
+				return false
+			}
+		}
+		return true
+	}
+	require.Eventually(t, isPruned, 1000*time.Second, 10*time.Millisecond, "expected error when loading pruned heights")
 }
 
 func TestMultiStore_PruningRestart(t *testing.T) {
@@ -618,10 +623,42 @@ func TestMultiStore_PruningRestart(t *testing.T) {
 	actualHeightToPrune = ms.pruningManager.GetPruningHeight(ms.LatestVersion())
 	require.Equal(t, int64(8), actualHeightToPrune)
 
-	for v := int64(1); v <= actualHeightToPrune; v++ {
-		_, err := ms.CacheMultiStoreWithVersion(v)
-		require.Error(t, err, "expected error when loading height: %d", v)
+	// Ensure async pruning is done
+	isPruned := func() bool {
+		ms.Commit() // to flush the batch with the pruned heights
+		for v := int64(1); v <= actualHeightToPrune; v++ {
+			if _, err := ms.CacheMultiStoreWithVersion(v); err == nil {
+				return false
+			}
+		}
+		return true
 	}
+
+	require.Eventually(t, isPruned, 1*time.Second, 10*time.Millisecond, "expected error when loading pruned heights")
+}
+
+var _ types.PausablePruner = &pauseableCommitKVStoreStub{}
+
+type pauseableCommitKVStoreStub struct {
+	types.CommitKVStore
+	pauseCalled []bool
+}
+
+func (p *pauseableCommitKVStoreStub) PausePruning(b bool) {
+	p.pauseCalled = append(p.pauseCalled, b)
+}
+
+func TestPausePruningOnCommit(t *testing.T) {
+	store := NewStore(dbm.NewMemDB(), log.NewNopLogger(), metrics.NewNoOpMetrics())
+	store.SetPruning(pruningtypes.NewCustomPruningOptions(2, 11))
+	store.MountStoreWithDB(testStoreKey1, types.StoreTypeIAVL, nil)
+	require.NoError(t, store.LoadLatestVersion())
+	myStub := &pauseableCommitKVStoreStub{CommitKVStore: store.stores[testStoreKey1].(types.CommitKVStore)}
+	store.stores[testStoreKey1] = myStub
+	// when
+	store.Commit()
+	// then
+	require.Equal(t, []bool{true, false}, myStub.pauseCalled)
 }
 
 // TestUnevenStoresHeightCheck tests if loading root store correctly errors when

store/types/store.go

@@ -29,6 +29,14 @@ type Committer interface {
 	GetPruning() pruningtypes.PruningOptions
 }
 
+type PausablePruner interface {
+	// PausePruning let the pruning handler know that the store is being committed
+	// or not, so the handler can decide to prune or not the store.
+	//
+	// NOTE: PausePruning(true) should be called before Commit() and PausePruning(false)
+	PausePruning(bool)
+}
+
 // Stores of MultiStore must implement CommitStore.
 type CommitStore interface {
 	Committer