Synopsis
This sections describes how the app-side mempool can be used and replaced.
v0.47 the application has its own mempool to allow much more granular
block building than previous versions. This change was enabled by
ABCI 1.0.
Notably it introduces the PrepareProposal and ProcessProposal steps of ABCI++.
Prepare Proposal
PrepareProposal handles construction of the block, meaning that when a proposer
is preparing to propose a block, it requests the application to evaluate a
RequestPrepareProposal, which contains a series of transactions from CometBFT’s
mempool. At this point, the application has complete control over the proposal.
It can modify, delete, and inject transactions from it’s own app-side mempool into
the proposal or even ignore all the transactions altogether. What the application
does with the transactions provided to it by RequestPrepareProposal have no
effect on CometBFT’s mempool.
Note, that the application defines the semantics of the PrepareProposal and it
MAY be non-deterministic and is only executed by the current block proposer.
Now, reading mempool twice in the previous sentence is confusing, lets break it down.
CometBFT has a mempool that handles gossiping transactions to other nodes
in the network. How these transactions are ordered is determined by CometBFT’s
mempool, typically FIFO. However, since the application is able to fully inspect
all transactions, it can provide greater control over transaction ordering.
Allowing the application to handle ordering enables the application to define how
it would like the block constructed.
Currently, there is a default PrepareProposal implementation provided by the application.
Copy
Ask AI
package baseapp
import (
"errors"
"fmt"
"sort"
"strings"
"github.com/cosmos/gogoproto/proto"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/crypto/tmhash"
"github.com/tendermint/tendermint/libs/log"
tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
dbm "github.com/tendermint/tm-db"
"golang.org/x/exp/maps"
codectypes "github.com/cosmos/cosmos-sdk/codec/types"
"github.com/cosmos/cosmos-sdk/snapshots"
"github.com/cosmos/cosmos-sdk/store"
"github.com/cosmos/cosmos-sdk/store/rootmulti"
storetypes "github.com/cosmos/cosmos-sdk/store/types"
sdk "github.com/cosmos/cosmos-sdk/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
"github.com/cosmos/cosmos-sdk/types/mempool"
)
const (
runTxModeCheck runTxMode = iota / Check a transaction
runTxModeReCheck / Recheck a (pending)
transaction after a commit
runTxModeSimulate / Simulate a transaction
runTxModeDeliver / Deliver a transaction
runTxPrepareProposal
runTxProcessProposal
)
var _ abci.Application = (*BaseApp)(nil)
type (
/ Enum mode for app.runTx
runTxMode uint8
/ StoreLoader defines a customizable function to control how we load the CommitMultiStore
/ from disk. This is useful for state migration, when loading a datastore written with
/ an older version of the software. In particular, if a module changed the substore key name
/ (or removed a substore)
between two versions of the software.
StoreLoader func(ms sdk.CommitMultiStore)
error
)
/ BaseApp reflects the ABCI application implementation.
type BaseApp struct { /nolint: maligned
/ initialized on creation
logger log.Logger
name string / application name from abci.Info
db dbm.DB / common DB backend
cms sdk.CommitMultiStore / Main (uncached)
state
qms sdk.MultiStore / Optional alternative multistore for querying only.
storeLoader StoreLoader / function to handle store loading, may be overridden with SetStoreLoader()
grpcQueryRouter *GRPCQueryRouter / router for redirecting gRPC query calls
msgServiceRouter *MsgServiceRouter / router for redirecting Msg service messages
interfaceRegistry codectypes.InterfaceRegistry
txDecoder sdk.TxDecoder / unmarshal []byte into sdk.Tx
txEncoder sdk.TxEncoder / marshal sdk.Tx into []byte
mempool mempool.Mempool / application side mempool
anteHandler sdk.AnteHandler / ante handler for fee and auth
postHandler sdk.AnteHandler / post handler, optional, e.g. for tips
initChainer sdk.InitChainer / initialize state with validators and state blob
beginBlocker sdk.BeginBlocker / logic to run before any txs
processProposal sdk.ProcessProposalHandler / the handler which runs on ABCI ProcessProposal
prepareProposal sdk.PrepareProposalHandler / the handler which runs on ABCI PrepareProposal
endBlocker sdk.EndBlocker / logic to run after all txs, and to determine valset changes
addrPeerFilter sdk.PeerFilter / filter peers by address and port
idPeerFilter sdk.PeerFilter / filter peers by node ID
fauxMerkleMode bool / if true, IAVL MountStores uses MountStoresDB for simulation speed.
/ manages snapshots, i.e. dumps of app state at certain intervals
snapshotManager *snapshots.Manager
/ volatile states:
/
/ checkState is set on InitChain and reset on Commit
/ deliverState is set on InitChain and BeginBlock and set to nil on Commit
checkState *state / for CheckTx
deliverState *state / for DeliverTx
processProposalState *state / for ProcessProposal
prepareProposalState *state / for PrepareProposal
/ an inter-block write-through cache provided to the context during deliverState
interBlockCache sdk.MultiStorePersistentCache
/ absent validators from begin block
voteInfos []abci.VoteInfo
/ paramStore is used to query for ABCI consensus parameters from an
/ application parameter store.
paramStore ParamStore
/ The minimum gas prices a validator is willing to accept for processing a
/ transaction. This is mainly used for DoS and spam prevention.
minGasPrices sdk.DecCoins
/ initialHeight is the initial height at which we start the baseapp
initialHeight int64
/ flag for sealing options and parameters to a BaseApp
sealed bool
/ block height at which to halt the chain and gracefully shutdown
haltHeight uint64
/ minimum block time (in Unix seconds)
at which to halt the chain and gracefully shutdown
haltTime uint64
/ minRetainBlocks defines the minimum block height offset from the current
/ block being committed, such that all blocks past this offset are pruned
/ from Tendermint. It is used as part of the process of determining the
/ ResponseCommit.RetainHeight value during ABCI Commit. A value of 0 indicates
/ that no blocks should be pruned.
/
/ Note: Tendermint block pruning is dependant on this parameter in conunction
/ with the unbonding (safety threshold)
period, state pruning and state sync
/ snapshot parameters to determine the correct minimum value of
/ ResponseCommit.RetainHeight.
minRetainBlocks uint64
/ application's version string
version string
/ application's protocol version that increments on every upgrade
/ if BaseApp is passed to the upgrade keeper's NewKeeper method.
appVersion uint64
/ recovery handler for app.runTx method
runTxRecoveryMiddleware recoveryMiddleware
/ trace set will return full stack traces for errors in ABCI Log field
trace bool
/ indexEvents defines the set of events in the form {
eventType
}.{
attributeKey
},
/ which informs Tendermint what to index. If empty, all events will be indexed.
indexEvents map[string]struct{
}
/ abciListeners for hooking into the ABCI message processing of the BaseApp
/ and exposing the requests and responses to external consumers
abciListeners []ABCIListener
}
/ NewBaseApp returns a reference to an initialized BaseApp. It accepts a
/ variadic number of option functions, which act on the BaseApp to set
/ configuration choices.
/
/ NOTE: The db is used to store the version number for now.
func NewBaseApp(
name string, logger log.Logger, db dbm.DB, txDecoder sdk.TxDecoder, options ...func(*BaseApp),
) *BaseApp {
app := &BaseApp{
logger: logger,
name: name,
db: db,
cms: store.NewCommitMultiStore(db),
storeLoader: DefaultStoreLoader,
grpcQueryRouter: NewGRPCQueryRouter(),
msgServiceRouter: NewMsgServiceRouter(),
txDecoder: txDecoder,
fauxMerkleMode: false,
}
for _, option := range options {
option(app)
}
if app.mempool == nil {
app.SetMempool(mempool.NoOpMempool{
})
}
if app.processProposal == nil {
app.SetProcessProposal(app.DefaultProcessProposal())
}
if app.prepareProposal == nil {
app.SetPrepareProposal(app.DefaultPrepareProposal())
}
if app.interBlockCache != nil {
app.cms.SetInterBlockCache(app.interBlockCache)
}
app.runTxRecoveryMiddleware = newDefaultRecoveryMiddleware()
return app
}
/ Name returns the name of the BaseApp.
func (app *BaseApp)
Name()
string {
return app.name
}
/ AppVersion returns the application's protocol version.
func (app *BaseApp)
AppVersion()
uint64 {
return app.appVersion
}
/ Version returns the application's version string.
func (app *BaseApp)
Version()
string {
return app.version
}
/ Logger returns the logger of the BaseApp.
func (app *BaseApp)
Logger()
log.Logger {
return app.logger
}
/ Trace returns the boolean value for logging error stack traces.
func (app *BaseApp)
Trace()
bool {
return app.trace
}
/ MsgServiceRouter returns the MsgServiceRouter of a BaseApp.
func (app *BaseApp)
MsgServiceRouter() *MsgServiceRouter {
return app.msgServiceRouter
}
/ SetMsgServiceRouter sets the MsgServiceRouter of a BaseApp.
func (app *BaseApp)
SetMsgServiceRouter(msgServiceRouter *MsgServiceRouter) {
app.msgServiceRouter = msgServiceRouter
}
/ MountStores mounts all IAVL or DB stores to the provided keys in the BaseApp
/ multistore.
func (app *BaseApp)
MountStores(keys ...storetypes.StoreKey) {
for _, key := range keys {
switch key.(type) {
case *storetypes.KVStoreKey:
if !app.fauxMerkleMode {
app.MountStore(key, storetypes.StoreTypeIAVL)
}
else {
/ StoreTypeDB doesn't do anything upon commit, and it doesn't
/ retain history, but it's useful for faster simulation.
app.MountStore(key, storetypes.StoreTypeDB)
}
case *storetypes.TransientStoreKey:
app.MountStore(key, storetypes.StoreTypeTransient)
case *storetypes.MemoryStoreKey:
app.MountStore(key, storetypes.StoreTypeMemory)
default:
panic(fmt.Sprintf("Unrecognized store key type :%T", key))
}
}
}
/ MountKVStores mounts all IAVL or DB stores to the provided keys in the
/ BaseApp multistore.
func (app *BaseApp)
MountKVStores(keys map[string]*storetypes.KVStoreKey) {
for _, key := range keys {
if !app.fauxMerkleMode {
app.MountStore(key, storetypes.StoreTypeIAVL)
}
else {
/ StoreTypeDB doesn't do anything upon commit, and it doesn't
/ retain history, but it's useful for faster simulation.
app.MountStore(key, storetypes.StoreTypeDB)
}
}
}
/ MountTransientStores mounts all transient stores to the provided keys in
/ the BaseApp multistore.
func (app *BaseApp)
MountTransientStores(keys map[string]*storetypes.TransientStoreKey) {
for _, key := range keys {
app.MountStore(key, storetypes.StoreTypeTransient)
}
}
/ MountMemoryStores mounts all in-memory KVStores with the BaseApp's internal
/ commit multi-store.
func (app *BaseApp)
MountMemoryStores(keys map[string]*storetypes.MemoryStoreKey) {
skeys := maps.Keys(keys)
sort.Strings(skeys)
for _, key := range skeys {
memKey := keys[key]
app.MountStore(memKey, storetypes.StoreTypeMemory)
}
}
/ MountStore mounts a store to the provided key in the BaseApp multistore,
/ using the default DB.
func (app *BaseApp)
MountStore(key storetypes.StoreKey, typ storetypes.StoreType) {
app.cms.MountStoreWithDB(key, typ, nil)
}
/ LoadLatestVersion loads the latest application version. It will panic if
/ called more than once on a running BaseApp.
func (app *BaseApp)
LoadLatestVersion()
error {
err := app.storeLoader(app.cms)
if err != nil {
return fmt.Errorf("failed to load latest version: %w", err)
}
return app.Init()
}
/ DefaultStoreLoader will be used by default and loads the latest version
func DefaultStoreLoader(ms sdk.CommitMultiStore)
error {
return ms.LoadLatestVersion()
}
/ CommitMultiStore returns the root multi-store.
/ App constructor can use this to access the `cms`.
/ UNSAFE: must not be used during the abci life cycle.
func (app *BaseApp)
CommitMultiStore()
sdk.CommitMultiStore {
return app.cms
}
/ SnapshotManager returns the snapshot manager.
/ application use this to register extra extension snapshotters.
func (app *BaseApp)
SnapshotManager() *snapshots.Manager {
return app.snapshotManager
}
/ LoadVersion loads the BaseApp application version. It will panic if called
/ more than once on a running baseapp.
func (app *BaseApp)
LoadVersion(version int64)
error {
app.logger.Info("NOTICE: this could take a long time to migrate IAVL store to fastnode if you enable Fast Node.\n")
err := app.cms.LoadVersion(version)
if err != nil {
return fmt.Errorf("failed to load version %d: %w", version, err)
}
return app.Init()
}
/ LastCommitID returns the last CommitID of the multistore.
func (app *BaseApp)
LastCommitID()
storetypes.CommitID {
return app.cms.LastCommitID()
}
/ LastBlockHeight returns the last committed block height.
func (app *BaseApp)
LastBlockHeight()
int64 {
return app.cms.LastCommitID().Version
}
/ Init initializes the app. It seals the app, preventing any
/ further modifications. In addition, it validates the app against
/ the earlier provided settings. Returns an error if validation fails.
/ nil otherwise. Panics if the app is already sealed.
func (app *BaseApp)
Init()
error {
if app.sealed {
panic("cannot call initFromMainStore: baseapp already sealed")
}
emptyHeader := tmproto.Header{
}
/ needed for the export command which inits from store but never calls initchain
app.setState(runTxModeCheck, emptyHeader)
/ needed for ABCI Replay Blocks mode which calls Prepare/Process proposal (InitChain is not called)
app.setState(runTxPrepareProposal, emptyHeader)
app.setState(runTxProcessProposal, emptyHeader)
app.Seal()
rms, ok := app.cms.(*rootmulti.Store)
if !ok {
return fmt.Errorf("invalid commit multi-store; expected %T, got: %T", &rootmulti.Store{
}, app.cms)
}
return rms.GetPruning().Validate()
}
func (app *BaseApp)
setMinGasPrices(gasPrices sdk.DecCoins) {
app.minGasPrices = gasPrices
}
func (app *BaseApp)
setHaltHeight(haltHeight uint64) {
app.haltHeight = haltHeight
}
func (app *BaseApp)
setHaltTime(haltTime uint64) {
app.haltTime = haltTime
}
func (app *BaseApp)
setMinRetainBlocks(minRetainBlocks uint64) {
app.minRetainBlocks = minRetainBlocks
}
func (app *BaseApp)
setInterBlockCache(cache sdk.MultiStorePersistentCache) {
app.interBlockCache = cache
}
func (app *BaseApp)
setTrace(trace bool) {
app.trace = trace
}
func (app *BaseApp)
setIndexEvents(ie []string) {
app.indexEvents = make(map[string]struct{
})
for _, e := range ie {
app.indexEvents[e] = struct{
}{
}
}
}
/ Seal seals a BaseApp. It prohibits any further modifications to a BaseApp.
func (app *BaseApp)
Seal() {
app.sealed = true
}
/ IsSealed returns true if the BaseApp is sealed and false otherwise.
func (app *BaseApp)
IsSealed()
bool {
return app.sealed
}
/ setState sets the BaseApp's state for the corresponding mode with a branched
/ multi-store (i.e. a CacheMultiStore)
and a new Context with the same
/ multi-store branch, and provided header.
func (app *BaseApp)
setState(mode runTxMode, header tmproto.Header) {
ms := app.cms.CacheMultiStore()
baseState := &state{
ms: ms,
ctx: sdk.NewContext(ms, header, false, app.logger),
}
switch mode {
case runTxModeCheck:
/ Minimum gas prices are also set. It is set on InitChain and reset on Commit.
baseState.ctx = baseState.ctx.WithIsCheckTx(true).WithMinGasPrices(app.minGasPrices)
app.checkState = baseState
case runTxModeDeliver:
/ It is set on InitChain and BeginBlock and set to nil on Commit.
app.deliverState = baseState
case runTxPrepareProposal:
/ It is set on InitChain and Commit.
app.prepareProposalState = baseState
case runTxProcessProposal:
/ It is set on InitChain and Commit.
app.processProposalState = baseState
default:
panic(fmt.Sprintf("invalid runTxMode for setState: %d", mode))
}
}
/ GetConsensusParams returns the current consensus parameters from the BaseApp's
/ ParamStore. If the BaseApp has no ParamStore defined, nil is returned.
func (app *BaseApp)
GetConsensusParams(ctx sdk.Context) *tmproto.ConsensusParams {
if app.paramStore == nil {
return nil
}
cp, err := app.paramStore.Get(ctx)
if err != nil {
panic(err)
}
return cp
}
/ StoreConsensusParams sets the consensus parameters to the baseapp's param store.
func (app *BaseApp)
StoreConsensusParams(ctx sdk.Context, cp *tmproto.ConsensusParams) {
if app.paramStore == nil {
panic("cannot store consensus params with no params store set")
}
if cp == nil {
return
}
app.paramStore.Set(ctx, cp)
/ We're explicitly not storing the Tendermint app_version in the param store. It's
/ stored instead in the x/upgrade store, with its own bump logic.
}
/ AddRunTxRecoveryHandler adds custom app.runTx method panic handlers.
func (app *BaseApp)
AddRunTxRecoveryHandler(handlers ...RecoveryHandler) {
for _, h := range handlers {
app.runTxRecoveryMiddleware = newRecoveryMiddleware(h, app.runTxRecoveryMiddleware)
}
}
/ GetMaximumBlockGas gets the maximum gas from the consensus params. It panics
/ if maximum block gas is less than negative one and returns zero if negative
/ one.
func (app *BaseApp)
GetMaximumBlockGas(ctx sdk.Context)
uint64 {
cp := app.GetConsensusParams(ctx)
if cp == nil || cp.Block == nil {
return 0
}
maxGas := cp.Block.MaxGas
switch {
case maxGas < -1:
panic(fmt.Sprintf("invalid maximum block gas: %d", maxGas))
case maxGas == -1:
return 0
default:
return uint64(maxGas)
}
}
func (app *BaseApp)
validateHeight(req abci.RequestBeginBlock)
error {
if req.Header.Height < 1 {
return fmt.Errorf("invalid height: %d", req.Header.Height)
}
/ expectedHeight holds the expected height to validate.
var expectedHeight int64
if app.LastBlockHeight() == 0 && app.initialHeight > 1 {
/ In this case, we're validating the first block of the chain (no
/ previous commit). The height we're expecting is the initial height.
expectedHeight = app.initialHeight
}
else {
/ This case can mean two things:
/ - either there was already a previous commit in the store, in which
/ case we increment the version from there,
/ - or there was no previous commit, and initial version was not set,
/ in which case we start at version 1.
expectedHeight = app.LastBlockHeight() + 1
}
if req.Header.Height != expectedHeight {
return fmt.Errorf("invalid height: %d; expected: %d", req.Header.Height, expectedHeight)
}
return nil
}
/ validateBasicTxMsgs executes basic validator calls for messages.
func validateBasicTxMsgs(msgs []sdk.Msg)
error {
if len(msgs) == 0 {
return sdkerrors.Wrap(sdkerrors.ErrInvalidRequest, "must contain at least one message")
}
for _, msg := range msgs {
err := msg.ValidateBasic()
if err != nil {
return err
}
}
return nil
}
/ Returns the application's deliverState if app is in runTxModeDeliver,
/ prepareProposalState if app is in runTxPrepareProposal, processProposalState
/ if app is in runTxProcessProposal, and checkState otherwise.
func (app *BaseApp)
getState(mode runTxMode) *state {
switch mode {
case runTxModeDeliver:
return app.deliverState
case runTxPrepareProposal:
return app.prepareProposalState
case runTxProcessProposal:
return app.processProposalState
default:
return app.checkState
}
}
/ retrieve the context for the tx w/ txBytes and other memoized values.
func (app *BaseApp)
getContextForTx(mode runTxMode, txBytes []byte)
sdk.Context {
modeState := app.getState(mode)
if modeState == nil {
panic(fmt.Sprintf("state is nil for mode %v", mode))
}
ctx := modeState.ctx.
WithTxBytes(txBytes).
WithVoteInfos(app.voteInfos)
ctx = ctx.WithConsensusParams(app.GetConsensusParams(ctx))
if mode == runTxModeReCheck {
ctx = ctx.WithIsReCheckTx(true)
}
if mode == runTxModeSimulate {
ctx, _ = ctx.CacheContext()
}
return ctx
}
/ cacheTxContext returns a new context based off of the provided context with
/ a branched multi-store.
func (app *BaseApp)
cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context, sdk.CacheMultiStore) {
ms := ctx.MultiStore()
/ TODO: https://github.com/cosmos/cosmos-sdk/issues/2824
msCache := ms.CacheMultiStore()
if msCache.TracingEnabled() {
msCache = msCache.SetTracingContext(
sdk.TraceContext(
map[string]interface{
}{
"txHash": fmt.Sprintf("%X", tmhash.Sum(txBytes)),
},
),
).(sdk.CacheMultiStore)
}
return ctx.WithMultiStore(msCache), msCache
}
/ runTx processes a transaction within a given execution mode, encoded transaction
/ bytes, and the decoded transaction itself. All state transitions occur through
/ a cached Context depending on the mode provided. State only gets persisted
/ if all messages get executed successfully and the execution mode is DeliverTx.
/ Note, gas execution info is always returned. A reference to a Result is
/ returned if the tx does not run out of gas and if all the messages are valid
/ and execute successfully. An error is returned otherwise.
func (app *BaseApp)
runTx(mode runTxMode, txBytes []byte) (gInfo sdk.GasInfo, result *sdk.Result, anteEvents []abci.Event, priority int64, err error) {
/ NOTE: GasWanted should be returned by the AnteHandler. GasUsed is
/ determined by the GasMeter. We need access to the context to get the gas
/ meter, so we initialize upfront.
var gasWanted uint64
ctx := app.getContextForTx(mode, txBytes)
ms := ctx.MultiStore()
/ only run the tx if there is block gas remaining
if mode == runTxModeDeliver && ctx.BlockGasMeter().IsOutOfGas() {
return gInfo, nil, nil, 0, sdkerrors.Wrap(sdkerrors.ErrOutOfGas, "no block gas left to run tx")
}
defer func() {
if r := recover(); r != nil {
recoveryMW := newOutOfGasRecoveryMiddleware(gasWanted, ctx, app.runTxRecoveryMiddleware)
err, result = processRecovery(r, recoveryMW), nil
}
gInfo = sdk.GasInfo{
GasWanted: gasWanted,
GasUsed: ctx.GasMeter().GasConsumed()
}
}()
blockGasConsumed := false
/ consumeBlockGas makes sure block gas is consumed at most once. It must happen after
/ tx processing, and must be executed even if tx processing fails. Hence, we use trick with `defer`
consumeBlockGas := func() {
if !blockGasConsumed {
blockGasConsumed = true
ctx.BlockGasMeter().ConsumeGas(
ctx.GasMeter().GasConsumedToLimit(), "block gas meter",
)
}
}
/ If BlockGasMeter()
panics it will be caught by the above recover and will
/ return an error - in any case BlockGasMeter will consume gas past the limit.
/
/ NOTE: This must exist in a separate defer function for the above recovery
/ to recover from this one.
if mode == runTxModeDeliver {
defer consumeBlockGas()
}
tx, err := app.txDecoder(txBytes)
if err != nil {
return sdk.GasInfo{
}, nil, nil, 0, err
}
msgs := tx.GetMsgs()
if err := validateBasicTxMsgs(msgs); err != nil {
return sdk.GasInfo{
}, nil, nil, 0, err
}
if app.anteHandler != nil {
var (
anteCtx sdk.Context
msCache sdk.CacheMultiStore
)
/ Branch context before AnteHandler call in case it aborts.
/ This is required for both CheckTx and DeliverTx.
/ Ref: https://github.com/cosmos/cosmos-sdk/issues/2772
/
/ NOTE: Alternatively, we could require that AnteHandler ensures that
/ writes do not happen if aborted/failed. This may have some
/ performance benefits, but it'll be more difficult to get right.
anteCtx, msCache = app.cacheTxContext(ctx, txBytes)
anteCtx = anteCtx.WithEventManager(sdk.NewEventManager())
newCtx, err := app.anteHandler(anteCtx, tx, mode == runTxModeSimulate)
if !newCtx.IsZero() {
/ At this point, newCtx.MultiStore()
is a store branch, or something else
/ replaced by the AnteHandler. We want the original multistore.
/
/ Also, in the case of the tx aborting, we need to track gas consumed via
/ the instantiated gas meter in the AnteHandler, so we update the context
/ prior to returning.
ctx = newCtx.WithMultiStore(ms)
}
events := ctx.EventManager().Events()
/ GasMeter expected to be set in AnteHandler
gasWanted = ctx.GasMeter().Limit()
if err != nil {
return gInfo, nil, nil, 0, err
}
priority = ctx.Priority()
msCache.Write()
anteEvents = events.ToABCIEvents()
}
if mode == runTxModeCheck {
err = app.mempool.Insert(ctx, tx)
if err != nil {
return gInfo, nil, anteEvents, priority, err
}
}
else if mode == runTxModeDeliver {
err = app.mempool.Remove(tx)
if err != nil && !errors.Is(err, mempool.ErrTxNotFound) {
return gInfo, nil, anteEvents, priority,
fmt.Errorf("failed to remove tx from mempool: %w", err)
}
}
/ Create a new Context based off of the existing Context with a MultiStore branch
/ in case message processing fails. At this point, the MultiStore
/ is a branch of a branch.
runMsgCtx, msCache := app.cacheTxContext(ctx, txBytes)
/ Attempt to execute all messages and only update state if all messages pass
/ and we're in DeliverTx. Note, runMsgs will never return a reference to a
/ Result if any single message fails or does not have a registered Handler.
result, err = app.runMsgs(runMsgCtx, msgs, mode)
if err == nil {
/ Run optional postHandlers.
/
/ Note: If the postHandler fails, we also revert the runMsgs state.
if app.postHandler != nil {
/ The runMsgCtx context currently contains events emitted by the ante handler.
/ We clear this to correctly order events without duplicates.
/ Note that the state is still preserved.
postCtx := runMsgCtx.WithEventManager(sdk.NewEventManager())
newCtx, err := app.postHandler(postCtx, tx, mode == runTxModeSimulate)
if err != nil {
return gInfo, nil, anteEvents, priority, err
}
result.Events = append(result.Events, newCtx.EventManager().ABCIEvents()...)
}
if mode == runTxModeDeliver {
/ When block gas exceeds, it'll panic and won't commit the cached store.
consumeBlockGas()
msCache.Write()
}
if len(anteEvents) > 0 && (mode == runTxModeDeliver || mode == runTxModeSimulate) {
/ append the events in the order of occurrence
result.Events = append(anteEvents, result.Events...)
}
}
return gInfo, result, anteEvents, priority, err
}
/ runMsgs iterates through a list of messages and executes them with the provided
/ Context and execution mode. Messages will only be executed during simulation
/ and DeliverTx. An error is returned if any single message fails or if a
/ Handler does not exist for a given message route. Otherwise, a reference to a
/ Result is returned. The caller must not commit state if an error is returned.
func (app *BaseApp)
runMsgs(ctx sdk.Context, msgs []sdk.Msg, mode runTxMode) (*sdk.Result, error) {
msgLogs := make(sdk.ABCIMessageLogs, 0, len(msgs))
events := sdk.EmptyEvents()
var msgResponses []*codectypes.Any
/ NOTE: GasWanted is determined by the AnteHandler and GasUsed by the GasMeter.
for i, msg := range msgs {
if mode != runTxModeDeliver && mode != runTxModeSimulate {
break
}
handler := app.msgServiceRouter.Handler(msg)
if handler == nil {
return nil, sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "can't route message %+v", msg)
}
/ ADR 031 request type routing
msgResult, err := handler(ctx, msg)
if err != nil {
return nil, sdkerrors.Wrapf(err, "failed to execute message; message index: %d", i)
}
/ create message events
msgEvents := createEvents(msgResult.GetEvents(), msg)
/ append message events, data and logs
/
/ Note: Each message result's data must be length-prefixed in order to
/ separate each result.
events = events.AppendEvents(msgEvents)
/ Each individual sdk.Result that went through the MsgServiceRouter
/ (which should represent 99% of the Msgs now, since everyone should
/ be using protobuf Msgs)
has exactly one Msg response, set inside
/ `WrapServiceResult`. We take that Msg response, and aggregate it
/ into an array.
if len(msgResult.MsgResponses) > 0 {
msgResponse := msgResult.MsgResponses[0]
if msgResponse == nil {
return nil, sdkerrors.ErrLogic.Wrapf("got nil Msg response at index %d for msg %s", i, sdk.MsgTypeURL(msg))
}
msgResponses = append(msgResponses, msgResponse)
}
msgLogs = append(msgLogs, sdk.NewABCIMessageLog(uint32(i), msgResult.Log, msgEvents))
}
data, err := makeABCIData(msgResponses)
if err != nil {
return nil, sdkerrors.Wrap(err, "failed to marshal tx data")
}
return &sdk.Result{
Data: data,
Log: strings.TrimSpace(msgLogs.String()),
Events: events.ToABCIEvents(),
MsgResponses: msgResponses,
}, nil
}
/ makeABCIData generates the Data field to be sent to ABCI Check/DeliverTx.
func makeABCIData(msgResponses []*codectypes.Any) ([]byte, error) {
return proto.Marshal(&sdk.TxMsgData{
MsgResponses: msgResponses
})
}
func createEvents(events sdk.Events, msg sdk.Msg)
sdk.Events {
eventMsgName := sdk.MsgTypeURL(msg)
msgEvent := sdk.NewEvent(sdk.EventTypeMessage, sdk.NewAttribute(sdk.AttributeKeyAction, eventMsgName))
/ we set the signer attribute as the sender
if len(msg.GetSigners()) > 0 && !msg.GetSigners()[0].Empty() {
msgEvent = msgEvent.AppendAttributes(sdk.NewAttribute(sdk.AttributeKeySender, msg.GetSigners()[0].String()))
}
/ verify that events have no module attribute set
if _, found := events.GetAttributes(sdk.AttributeKeyModule); !found {
/ here we assume that routes module name is the second element of the route
/ e.g. "cosmos.bank.v1beta1.MsgSend" => "bank"
moduleName := strings.Split(eventMsgName, ".")
if len(moduleName) > 1 {
msgEvent = msgEvent.AppendAttributes(sdk.NewAttribute(sdk.AttributeKeyModule, moduleName[1]))
}
}
return sdk.Events{
msgEvent
}.AppendEvents(events)
}
/ DefaultPrepareProposal returns the default implementation for processing an
/ ABCI proposal. The application's mempool is enumerated and all valid
/ transactions are added to the proposal. Transactions are valid if they:
/
/ 1)
Successfully encode to bytes.
/ 2)
Are valid (i.e. pass runTx, AnteHandler only).
/
/ Enumeration is halted once RequestPrepareProposal.MaxBytes of transactions is
/ reached or the mempool is exhausted.
/
/ Note:
/
/ - Step (2)
is identical to the validation step performed in
/ DefaultProcessProposal. It is very important that the same validation logic
/ is used in both steps, and applications must ensure that this is the case in
/ non-default handlers.
/
/ - If no mempool is set or if the mempool is a no-op mempool, the transactions
/ requested from Tendermint will simply be returned, which, by default, are in
/ FIFO order.
func (app *BaseApp)
DefaultPrepareProposal()
sdk.PrepareProposalHandler {
return func(ctx sdk.Context, req abci.RequestPrepareProposal)
abci.ResponsePrepareProposal {
/ If the mempool is nil or a no-op mempool, we simply return the transactions
/ requested from Tendermint, which, by default, should be in FIFO order.
_, isNoOp := app.mempool.(mempool.NoOpMempool)
if app.mempool == nil || isNoOp {
return abci.ResponsePrepareProposal{
Txs: req.Txs
}
}
var (
txsBytes [][]byte
byteCount int64
)
iterator := app.mempool.Select(ctx, req.Txs)
for iterator != nil {
memTx := iterator.Tx()
bz, err := app.txEncoder(memTx)
if err != nil {
panic(err)
}
txSize := int64(len(bz))
/ NOTE: Since runTx was already executed in CheckTx, which calls
/ mempool.Insert, ideally everything in the pool should be valid. But
/ some mempool implementations may insert invalid txs, so we check again.
_, _, _, _, err = app.runTx(runTxPrepareProposal, bz)
if err != nil {
err := app.mempool.Remove(memTx)
if err != nil && !errors.Is(err, mempool.ErrTxNotFound) {
panic(err)
}
iterator = iterator.Next()
continue
}
else if byteCount += txSize; byteCount <= req.MaxTxBytes {
txsBytes = append(txsBytes, bz)
}
else {
break
}
iterator = iterator.Next()
}
return abci.ResponsePrepareProposal{
Txs: txsBytes
}
}
}
/ DefaultProcessProposal returns the default implementation for processing an ABCI proposal.
/ Every transaction in the proposal must pass 2 conditions:
/
/ 1. The transaction bytes must decode to a valid transaction.
/ 2. The transaction must be valid (i.e. pass runTx, AnteHandler only)
/
/ If any transaction fails to pass either condition, the proposal is rejected. Note that step (2)
is identical to the
/ validation step performed in DefaultPrepareProposal. It is very important that the same validation logic is used
/ in both steps, and applications must ensure that this is the case in non-default handlers.
func (app *BaseApp)
DefaultProcessProposal()
sdk.ProcessProposalHandler {
return func(ctx sdk.Context, req abci.RequestProcessProposal)
abci.ResponseProcessProposal {
for _, txBytes := range req.Txs {
_, err := app.txDecoder(txBytes)
if err != nil {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_REJECT
}
}
_, _, _, _, err = app.runTx(runTxProcessProposal, txBytes)
if err != nil {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_REJECT
}
}
}
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_ACCEPT
}
}
}
/ NoOpPrepareProposal defines a no-op PrepareProposal handler. It will always
/ return the transactions sent by the client's request.
func NoOpPrepareProposal()
sdk.PrepareProposalHandler {
return func(_ sdk.Context, req abci.RequestPrepareProposal)
abci.ResponsePrepareProposal {
return abci.ResponsePrepareProposal{
Txs: req.Txs
}
}
}
/ NoOpProcessProposal defines a no-op ProcessProposal Handler. It will always
/ return ACCEPT.
func NoOpProcessProposal()
sdk.ProcessProposalHandler {
return func(_ sdk.Context, _ abci.RequestProcessProposal)
abci.ResponseProcessProposal {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_ACCEPT
}
}
}
app.go:
Copy
Ask AI
prepareOpt := func(app *baseapp.BaseApp) {
abciPropHandler := baseapp.NewDefaultProposalHandler(mempool, app)
app.SetPrepareProposal(abciPropHandler.PrepareProposalHandler())
}
baseAppOptions = append(baseAppOptions, prepareOpt)
Process Proposal
ProcessProposal handles the validation of a proposal from PrepareProposal,
which also includes a block header. Meaning, that after a block has been proposed
the other validators have the right to vote on a block. The validator in the
default implementation of PrepareProposal runs basic validity checks on each
transaction.
Note, ProcessProposal MAY NOT be non-deterministic, i.e. it must be deterministic.
This means if ProcessProposal panics or fails and we reject, all honest validator
processes will prevote nil and the CometBFT round will proceed again until a valid
proposal is proposed.
Here is the implementation of the default implementation:
Copy
Ask AI
package baseapp
import (
"errors"
"fmt"
"sort"
"strings"
"github.com/cosmos/gogoproto/proto"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/crypto/tmhash"
"github.com/tendermint/tendermint/libs/log"
tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
dbm "github.com/tendermint/tm-db"
"golang.org/x/exp/maps"
codectypes "github.com/cosmos/cosmos-sdk/codec/types"
"github.com/cosmos/cosmos-sdk/snapshots"
"github.com/cosmos/cosmos-sdk/store"
"github.com/cosmos/cosmos-sdk/store/rootmulti"
storetypes "github.com/cosmos/cosmos-sdk/store/types"
sdk "github.com/cosmos/cosmos-sdk/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
"github.com/cosmos/cosmos-sdk/types/mempool"
)
const (
runTxModeCheck runTxMode = iota / Check a transaction
runTxModeReCheck / Recheck a (pending)
transaction after a commit
runTxModeSimulate / Simulate a transaction
runTxModeDeliver / Deliver a transaction
runTxPrepareProposal
runTxProcessProposal
)
var _ abci.Application = (*BaseApp)(nil)
type (
/ Enum mode for app.runTx
runTxMode uint8
/ StoreLoader defines a customizable function to control how we load the CommitMultiStore
/ from disk. This is useful for state migration, when loading a datastore written with
/ an older version of the software. In particular, if a module changed the substore key name
/ (or removed a substore)
between two versions of the software.
StoreLoader func(ms sdk.CommitMultiStore)
error
)
/ BaseApp reflects the ABCI application implementation.
type BaseApp struct { /nolint: maligned
/ initialized on creation
logger log.Logger
name string / application name from abci.Info
db dbm.DB / common DB backend
cms sdk.CommitMultiStore / Main (uncached)
state
qms sdk.MultiStore / Optional alternative multistore for querying only.
storeLoader StoreLoader / function to handle store loading, may be overridden with SetStoreLoader()
grpcQueryRouter *GRPCQueryRouter / router for redirecting gRPC query calls
msgServiceRouter *MsgServiceRouter / router for redirecting Msg service messages
interfaceRegistry codectypes.InterfaceRegistry
txDecoder sdk.TxDecoder / unmarshal []byte into sdk.Tx
txEncoder sdk.TxEncoder / marshal sdk.Tx into []byte
mempool mempool.Mempool / application side mempool
anteHandler sdk.AnteHandler / ante handler for fee and auth
postHandler sdk.AnteHandler / post handler, optional, e.g. for tips
initChainer sdk.InitChainer / initialize state with validators and state blob
beginBlocker sdk.BeginBlocker / logic to run before any txs
processProposal sdk.ProcessProposalHandler / the handler which runs on ABCI ProcessProposal
prepareProposal sdk.PrepareProposalHandler / the handler which runs on ABCI PrepareProposal
endBlocker sdk.EndBlocker / logic to run after all txs, and to determine valset changes
addrPeerFilter sdk.PeerFilter / filter peers by address and port
idPeerFilter sdk.PeerFilter / filter peers by node ID
fauxMerkleMode bool / if true, IAVL MountStores uses MountStoresDB for simulation speed.
/ manages snapshots, i.e. dumps of app state at certain intervals
snapshotManager *snapshots.Manager
/ volatile states:
/
/ checkState is set on InitChain and reset on Commit
/ deliverState is set on InitChain and BeginBlock and set to nil on Commit
checkState *state / for CheckTx
deliverState *state / for DeliverTx
processProposalState *state / for ProcessProposal
prepareProposalState *state / for PrepareProposal
/ an inter-block write-through cache provided to the context during deliverState
interBlockCache sdk.MultiStorePersistentCache
/ absent validators from begin block
voteInfos []abci.VoteInfo
/ paramStore is used to query for ABCI consensus parameters from an
/ application parameter store.
paramStore ParamStore
/ The minimum gas prices a validator is willing to accept for processing a
/ transaction. This is mainly used for DoS and spam prevention.
minGasPrices sdk.DecCoins
/ initialHeight is the initial height at which we start the baseapp
initialHeight int64
/ flag for sealing options and parameters to a BaseApp
sealed bool
/ block height at which to halt the chain and gracefully shutdown
haltHeight uint64
/ minimum block time (in Unix seconds)
at which to halt the chain and gracefully shutdown
haltTime uint64
/ minRetainBlocks defines the minimum block height offset from the current
/ block being committed, such that all blocks past this offset are pruned
/ from Tendermint. It is used as part of the process of determining the
/ ResponseCommit.RetainHeight value during ABCI Commit. A value of 0 indicates
/ that no blocks should be pruned.
/
/ Note: Tendermint block pruning is dependant on this parameter in conunction
/ with the unbonding (safety threshold)
period, state pruning and state sync
/ snapshot parameters to determine the correct minimum value of
/ ResponseCommit.RetainHeight.
minRetainBlocks uint64
/ application's version string
version string
/ application's protocol version that increments on every upgrade
/ if BaseApp is passed to the upgrade keeper's NewKeeper method.
appVersion uint64
/ recovery handler for app.runTx method
runTxRecoveryMiddleware recoveryMiddleware
/ trace set will return full stack traces for errors in ABCI Log field
trace bool
/ indexEvents defines the set of events in the form {
eventType
}.{
attributeKey
},
/ which informs Tendermint what to index. If empty, all events will be indexed.
indexEvents map[string]struct{
}
/ abciListeners for hooking into the ABCI message processing of the BaseApp
/ and exposing the requests and responses to external consumers
abciListeners []ABCIListener
}
/ NewBaseApp returns a reference to an initialized BaseApp. It accepts a
/ variadic number of option functions, which act on the BaseApp to set
/ configuration choices.
/
/ NOTE: The db is used to store the version number for now.
func NewBaseApp(
name string, logger log.Logger, db dbm.DB, txDecoder sdk.TxDecoder, options ...func(*BaseApp),
) *BaseApp {
app := &BaseApp{
logger: logger,
name: name,
db: db,
cms: store.NewCommitMultiStore(db),
storeLoader: DefaultStoreLoader,
grpcQueryRouter: NewGRPCQueryRouter(),
msgServiceRouter: NewMsgServiceRouter(),
txDecoder: txDecoder,
fauxMerkleMode: false,
}
for _, option := range options {
option(app)
}
if app.mempool == nil {
app.SetMempool(mempool.NoOpMempool{
})
}
if app.processProposal == nil {
app.SetProcessProposal(app.DefaultProcessProposal())
}
if app.prepareProposal == nil {
app.SetPrepareProposal(app.DefaultPrepareProposal())
}
if app.interBlockCache != nil {
app.cms.SetInterBlockCache(app.interBlockCache)
}
app.runTxRecoveryMiddleware = newDefaultRecoveryMiddleware()
return app
}
/ Name returns the name of the BaseApp.
func (app *BaseApp)
Name()
string {
return app.name
}
/ AppVersion returns the application's protocol version.
func (app *BaseApp)
AppVersion()
uint64 {
return app.appVersion
}
/ Version returns the application's version string.
func (app *BaseApp)
Version()
string {
return app.version
}
/ Logger returns the logger of the BaseApp.
func (app *BaseApp)
Logger()
log.Logger {
return app.logger
}
/ Trace returns the boolean value for logging error stack traces.
func (app *BaseApp)
Trace()
bool {
return app.trace
}
/ MsgServiceRouter returns the MsgServiceRouter of a BaseApp.
func (app *BaseApp)
MsgServiceRouter() *MsgServiceRouter {
return app.msgServiceRouter
}
/ SetMsgServiceRouter sets the MsgServiceRouter of a BaseApp.
func (app *BaseApp)
SetMsgServiceRouter(msgServiceRouter *MsgServiceRouter) {
app.msgServiceRouter = msgServiceRouter
}
/ MountStores mounts all IAVL or DB stores to the provided keys in the BaseApp
/ multistore.
func (app *BaseApp)
MountStores(keys ...storetypes.StoreKey) {
for _, key := range keys {
switch key.(type) {
case *storetypes.KVStoreKey:
if !app.fauxMerkleMode {
app.MountStore(key, storetypes.StoreTypeIAVL)
}
else {
/ StoreTypeDB doesn't do anything upon commit, and it doesn't
/ retain history, but it's useful for faster simulation.
app.MountStore(key, storetypes.StoreTypeDB)
}
case *storetypes.TransientStoreKey:
app.MountStore(key, storetypes.StoreTypeTransient)
case *storetypes.MemoryStoreKey:
app.MountStore(key, storetypes.StoreTypeMemory)
default:
panic(fmt.Sprintf("Unrecognized store key type :%T", key))
}
}
}
/ MountKVStores mounts all IAVL or DB stores to the provided keys in the
/ BaseApp multistore.
func (app *BaseApp)
MountKVStores(keys map[string]*storetypes.KVStoreKey) {
for _, key := range keys {
if !app.fauxMerkleMode {
app.MountStore(key, storetypes.StoreTypeIAVL)
}
else {
/ StoreTypeDB doesn't do anything upon commit, and it doesn't
/ retain history, but it's useful for faster simulation.
app.MountStore(key, storetypes.StoreTypeDB)
}
}
}
/ MountTransientStores mounts all transient stores to the provided keys in
/ the BaseApp multistore.
func (app *BaseApp)
MountTransientStores(keys map[string]*storetypes.TransientStoreKey) {
for _, key := range keys {
app.MountStore(key, storetypes.StoreTypeTransient)
}
}
/ MountMemoryStores mounts all in-memory KVStores with the BaseApp's internal
/ commit multi-store.
func (app *BaseApp)
MountMemoryStores(keys map[string]*storetypes.MemoryStoreKey) {
skeys := maps.Keys(keys)
sort.Strings(skeys)
for _, key := range skeys {
memKey := keys[key]
app.MountStore(memKey, storetypes.StoreTypeMemory)
}
}
/ MountStore mounts a store to the provided key in the BaseApp multistore,
/ using the default DB.
func (app *BaseApp)
MountStore(key storetypes.StoreKey, typ storetypes.StoreType) {
app.cms.MountStoreWithDB(key, typ, nil)
}
/ LoadLatestVersion loads the latest application version. It will panic if
/ called more than once on a running BaseApp.
func (app *BaseApp)
LoadLatestVersion()
error {
err := app.storeLoader(app.cms)
if err != nil {
return fmt.Errorf("failed to load latest version: %w", err)
}
return app.Init()
}
/ DefaultStoreLoader will be used by default and loads the latest version
func DefaultStoreLoader(ms sdk.CommitMultiStore)
error {
return ms.LoadLatestVersion()
}
/ CommitMultiStore returns the root multi-store.
/ App constructor can use this to access the `cms`.
/ UNSAFE: must not be used during the abci life cycle.
func (app *BaseApp)
CommitMultiStore()
sdk.CommitMultiStore {
return app.cms
}
/ SnapshotManager returns the snapshot manager.
/ application use this to register extra extension snapshotters.
func (app *BaseApp)
SnapshotManager() *snapshots.Manager {
return app.snapshotManager
}
/ LoadVersion loads the BaseApp application version. It will panic if called
/ more than once on a running baseapp.
func (app *BaseApp)
LoadVersion(version int64)
error {
app.logger.Info("NOTICE: this could take a long time to migrate IAVL store to fastnode if you enable Fast Node.\n")
err := app.cms.LoadVersion(version)
if err != nil {
return fmt.Errorf("failed to load version %d: %w", version, err)
}
return app.Init()
}
/ LastCommitID returns the last CommitID of the multistore.
func (app *BaseApp)
LastCommitID()
storetypes.CommitID {
return app.cms.LastCommitID()
}
/ LastBlockHeight returns the last committed block height.
func (app *BaseApp)
LastBlockHeight()
int64 {
return app.cms.LastCommitID().Version
}
/ Init initializes the app. It seals the app, preventing any
/ further modifications. In addition, it validates the app against
/ the earlier provided settings. Returns an error if validation fails.
/ nil otherwise. Panics if the app is already sealed.
func (app *BaseApp)
Init()
error {
if app.sealed {
panic("cannot call initFromMainStore: baseapp already sealed")
}
emptyHeader := tmproto.Header{
}
/ needed for the export command which inits from store but never calls initchain
app.setState(runTxModeCheck, emptyHeader)
/ needed for ABCI Replay Blocks mode which calls Prepare/Process proposal (InitChain is not called)
app.setState(runTxPrepareProposal, emptyHeader)
app.setState(runTxProcessProposal, emptyHeader)
app.Seal()
rms, ok := app.cms.(*rootmulti.Store)
if !ok {
return fmt.Errorf("invalid commit multi-store; expected %T, got: %T", &rootmulti.Store{
}, app.cms)
}
return rms.GetPruning().Validate()
}
func (app *BaseApp)
setMinGasPrices(gasPrices sdk.DecCoins) {
app.minGasPrices = gasPrices
}
func (app *BaseApp)
setHaltHeight(haltHeight uint64) {
app.haltHeight = haltHeight
}
func (app *BaseApp)
setHaltTime(haltTime uint64) {
app.haltTime = haltTime
}
func (app *BaseApp)
setMinRetainBlocks(minRetainBlocks uint64) {
app.minRetainBlocks = minRetainBlocks
}
func (app *BaseApp)
setInterBlockCache(cache sdk.MultiStorePersistentCache) {
app.interBlockCache = cache
}
func (app *BaseApp)
setTrace(trace bool) {
app.trace = trace
}
func (app *BaseApp)
setIndexEvents(ie []string) {
app.indexEvents = make(map[string]struct{
})
for _, e := range ie {
app.indexEvents[e] = struct{
}{
}
}
}
/ Seal seals a BaseApp. It prohibits any further modifications to a BaseApp.
func (app *BaseApp)
Seal() {
app.sealed = true
}
/ IsSealed returns true if the BaseApp is sealed and false otherwise.
func (app *BaseApp)
IsSealed()
bool {
return app.sealed
}
/ setState sets the BaseApp's state for the corresponding mode with a branched
/ multi-store (i.e. a CacheMultiStore)
and a new Context with the same
/ multi-store branch, and provided header.
func (app *BaseApp)
setState(mode runTxMode, header tmproto.Header) {
ms := app.cms.CacheMultiStore()
baseState := &state{
ms: ms,
ctx: sdk.NewContext(ms, header, false, app.logger),
}
switch mode {
case runTxModeCheck:
/ Minimum gas prices are also set. It is set on InitChain and reset on Commit.
baseState.ctx = baseState.ctx.WithIsCheckTx(true).WithMinGasPrices(app.minGasPrices)
app.checkState = baseState
case runTxModeDeliver:
/ It is set on InitChain and BeginBlock and set to nil on Commit.
app.deliverState = baseState
case runTxPrepareProposal:
/ It is set on InitChain and Commit.
app.prepareProposalState = baseState
case runTxProcessProposal:
/ It is set on InitChain and Commit.
app.processProposalState = baseState
default:
panic(fmt.Sprintf("invalid runTxMode for setState: %d", mode))
}
}
/ GetConsensusParams returns the current consensus parameters from the BaseApp's
/ ParamStore. If the BaseApp has no ParamStore defined, nil is returned.
func (app *BaseApp)
GetConsensusParams(ctx sdk.Context) *tmproto.ConsensusParams {
if app.paramStore == nil {
return nil
}
cp, err := app.paramStore.Get(ctx)
if err != nil {
panic(err)
}
return cp
}
/ StoreConsensusParams sets the consensus parameters to the baseapp's param store.
func (app *BaseApp)
StoreConsensusParams(ctx sdk.Context, cp *tmproto.ConsensusParams) {
if app.paramStore == nil {
panic("cannot store consensus params with no params store set")
}
if cp == nil {
return
}
app.paramStore.Set(ctx, cp)
/ We're explicitly not storing the Tendermint app_version in the param store. It's
/ stored instead in the x/upgrade store, with its own bump logic.
}
/ AddRunTxRecoveryHandler adds custom app.runTx method panic handlers.
func (app *BaseApp)
AddRunTxRecoveryHandler(handlers ...RecoveryHandler) {
for _, h := range handlers {
app.runTxRecoveryMiddleware = newRecoveryMiddleware(h, app.runTxRecoveryMiddleware)
}
}
/ GetMaximumBlockGas gets the maximum gas from the consensus params. It panics
/ if maximum block gas is less than negative one and returns zero if negative
/ one.
func (app *BaseApp)
GetMaximumBlockGas(ctx sdk.Context)
uint64 {
cp := app.GetConsensusParams(ctx)
if cp == nil || cp.Block == nil {
return 0
}
maxGas := cp.Block.MaxGas
switch {
case maxGas < -1:
panic(fmt.Sprintf("invalid maximum block gas: %d", maxGas))
case maxGas == -1:
return 0
default:
return uint64(maxGas)
}
}
func (app *BaseApp)
validateHeight(req abci.RequestBeginBlock)
error {
if req.Header.Height < 1 {
return fmt.Errorf("invalid height: %d", req.Header.Height)
}
/ expectedHeight holds the expected height to validate.
var expectedHeight int64
if app.LastBlockHeight() == 0 && app.initialHeight > 1 {
/ In this case, we're validating the first block of the chain (no
/ previous commit). The height we're expecting is the initial height.
expectedHeight = app.initialHeight
}
else {
/ This case can mean two things:
/ - either there was already a previous commit in the store, in which
/ case we increment the version from there,
/ - or there was no previous commit, and initial version was not set,
/ in which case we start at version 1.
expectedHeight = app.LastBlockHeight() + 1
}
if req.Header.Height != expectedHeight {
return fmt.Errorf("invalid height: %d; expected: %d", req.Header.Height, expectedHeight)
}
return nil
}
/ validateBasicTxMsgs executes basic validator calls for messages.
func validateBasicTxMsgs(msgs []sdk.Msg)
error {
if len(msgs) == 0 {
return sdkerrors.Wrap(sdkerrors.ErrInvalidRequest, "must contain at least one message")
}
for _, msg := range msgs {
err := msg.ValidateBasic()
if err != nil {
return err
}
}
return nil
}
/ Returns the application's deliverState if app is in runTxModeDeliver,
/ prepareProposalState if app is in runTxPrepareProposal, processProposalState
/ if app is in runTxProcessProposal, and checkState otherwise.
func (app *BaseApp)
getState(mode runTxMode) *state {
switch mode {
case runTxModeDeliver:
return app.deliverState
case runTxPrepareProposal:
return app.prepareProposalState
case runTxProcessProposal:
return app.processProposalState
default:
return app.checkState
}
}
/ retrieve the context for the tx w/ txBytes and other memoized values.
func (app *BaseApp)
getContextForTx(mode runTxMode, txBytes []byte)
sdk.Context {
modeState := app.getState(mode)
if modeState == nil {
panic(fmt.Sprintf("state is nil for mode %v", mode))
}
ctx := modeState.ctx.
WithTxBytes(txBytes).
WithVoteInfos(app.voteInfos)
ctx = ctx.WithConsensusParams(app.GetConsensusParams(ctx))
if mode == runTxModeReCheck {
ctx = ctx.WithIsReCheckTx(true)
}
if mode == runTxModeSimulate {
ctx, _ = ctx.CacheContext()
}
return ctx
}
/ cacheTxContext returns a new context based off of the provided context with
/ a branched multi-store.
func (app *BaseApp)
cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context, sdk.CacheMultiStore) {
ms := ctx.MultiStore()
/ TODO: https://github.com/cosmos/cosmos-sdk/issues/2824
msCache := ms.CacheMultiStore()
if msCache.TracingEnabled() {
msCache = msCache.SetTracingContext(
sdk.TraceContext(
map[string]interface{
}{
"txHash": fmt.Sprintf("%X", tmhash.Sum(txBytes)),
},
),
).(sdk.CacheMultiStore)
}
return ctx.WithMultiStore(msCache), msCache
}
/ runTx processes a transaction within a given execution mode, encoded transaction
/ bytes, and the decoded transaction itself. All state transitions occur through
/ a cached Context depending on the mode provided. State only gets persisted
/ if all messages get executed successfully and the execution mode is DeliverTx.
/ Note, gas execution info is always returned. A reference to a Result is
/ returned if the tx does not run out of gas and if all the messages are valid
/ and execute successfully. An error is returned otherwise.
func (app *BaseApp)
runTx(mode runTxMode, txBytes []byte) (gInfo sdk.GasInfo, result *sdk.Result, anteEvents []abci.Event, priority int64, err error) {
/ NOTE: GasWanted should be returned by the AnteHandler. GasUsed is
/ determined by the GasMeter. We need access to the context to get the gas
/ meter, so we initialize upfront.
var gasWanted uint64
ctx := app.getContextForTx(mode, txBytes)
ms := ctx.MultiStore()
/ only run the tx if there is block gas remaining
if mode == runTxModeDeliver && ctx.BlockGasMeter().IsOutOfGas() {
return gInfo, nil, nil, 0, sdkerrors.Wrap(sdkerrors.ErrOutOfGas, "no block gas left to run tx")
}
defer func() {
if r := recover(); r != nil {
recoveryMW := newOutOfGasRecoveryMiddleware(gasWanted, ctx, app.runTxRecoveryMiddleware)
err, result = processRecovery(r, recoveryMW), nil
}
gInfo = sdk.GasInfo{
GasWanted: gasWanted,
GasUsed: ctx.GasMeter().GasConsumed()
}
}()
blockGasConsumed := false
/ consumeBlockGas makes sure block gas is consumed at most once. It must happen after
/ tx processing, and must be executed even if tx processing fails. Hence, we use trick with `defer`
consumeBlockGas := func() {
if !blockGasConsumed {
blockGasConsumed = true
ctx.BlockGasMeter().ConsumeGas(
ctx.GasMeter().GasConsumedToLimit(), "block gas meter",
)
}
}
/ If BlockGasMeter()
panics it will be caught by the above recover and will
/ return an error - in any case BlockGasMeter will consume gas past the limit.
/
/ NOTE: This must exist in a separate defer function for the above recovery
/ to recover from this one.
if mode == runTxModeDeliver {
defer consumeBlockGas()
}
tx, err := app.txDecoder(txBytes)
if err != nil {
return sdk.GasInfo{
}, nil, nil, 0, err
}
msgs := tx.GetMsgs()
if err := validateBasicTxMsgs(msgs); err != nil {
return sdk.GasInfo{
}, nil, nil, 0, err
}
if app.anteHandler != nil {
var (
anteCtx sdk.Context
msCache sdk.CacheMultiStore
)
/ Branch context before AnteHandler call in case it aborts.
/ This is required for both CheckTx and DeliverTx.
/ Ref: https://github.com/cosmos/cosmos-sdk/issues/2772
/
/ NOTE: Alternatively, we could require that AnteHandler ensures that
/ writes do not happen if aborted/failed. This may have some
/ performance benefits, but it'll be more difficult to get right.
anteCtx, msCache = app.cacheTxContext(ctx, txBytes)
anteCtx = anteCtx.WithEventManager(sdk.NewEventManager())
newCtx, err := app.anteHandler(anteCtx, tx, mode == runTxModeSimulate)
if !newCtx.IsZero() {
/ At this point, newCtx.MultiStore()
is a store branch, or something else
/ replaced by the AnteHandler. We want the original multistore.
/
/ Also, in the case of the tx aborting, we need to track gas consumed via
/ the instantiated gas meter in the AnteHandler, so we update the context
/ prior to returning.
ctx = newCtx.WithMultiStore(ms)
}
events := ctx.EventManager().Events()
/ GasMeter expected to be set in AnteHandler
gasWanted = ctx.GasMeter().Limit()
if err != nil {
return gInfo, nil, nil, 0, err
}
priority = ctx.Priority()
msCache.Write()
anteEvents = events.ToABCIEvents()
}
if mode == runTxModeCheck {
err = app.mempool.Insert(ctx, tx)
if err != nil {
return gInfo, nil, anteEvents, priority, err
}
}
else if mode == runTxModeDeliver {
err = app.mempool.Remove(tx)
if err != nil && !errors.Is(err, mempool.ErrTxNotFound) {
return gInfo, nil, anteEvents, priority,
fmt.Errorf("failed to remove tx from mempool: %w", err)
}
}
/ Create a new Context based off of the existing Context with a MultiStore branch
/ in case message processing fails. At this point, the MultiStore
/ is a branch of a branch.
runMsgCtx, msCache := app.cacheTxContext(ctx, txBytes)
/ Attempt to execute all messages and only update state if all messages pass
/ and we're in DeliverTx. Note, runMsgs will never return a reference to a
/ Result if any single message fails or does not have a registered Handler.
result, err = app.runMsgs(runMsgCtx, msgs, mode)
if err == nil {
/ Run optional postHandlers.
/
/ Note: If the postHandler fails, we also revert the runMsgs state.
if app.postHandler != nil {
/ The runMsgCtx context currently contains events emitted by the ante handler.
/ We clear this to correctly order events without duplicates.
/ Note that the state is still preserved.
postCtx := runMsgCtx.WithEventManager(sdk.NewEventManager())
newCtx, err := app.postHandler(postCtx, tx, mode == runTxModeSimulate)
if err != nil {
return gInfo, nil, anteEvents, priority, err
}
result.Events = append(result.Events, newCtx.EventManager().ABCIEvents()...)
}
if mode == runTxModeDeliver {
/ When block gas exceeds, it'll panic and won't commit the cached store.
consumeBlockGas()
msCache.Write()
}
if len(anteEvents) > 0 && (mode == runTxModeDeliver || mode == runTxModeSimulate) {
/ append the events in the order of occurrence
result.Events = append(anteEvents, result.Events...)
}
}
return gInfo, result, anteEvents, priority, err
}
/ runMsgs iterates through a list of messages and executes them with the provided
/ Context and execution mode. Messages will only be executed during simulation
/ and DeliverTx. An error is returned if any single message fails or if a
/ Handler does not exist for a given message route. Otherwise, a reference to a
/ Result is returned. The caller must not commit state if an error is returned.
func (app *BaseApp)
runMsgs(ctx sdk.Context, msgs []sdk.Msg, mode runTxMode) (*sdk.Result, error) {
msgLogs := make(sdk.ABCIMessageLogs, 0, len(msgs))
events := sdk.EmptyEvents()
var msgResponses []*codectypes.Any
/ NOTE: GasWanted is determined by the AnteHandler and GasUsed by the GasMeter.
for i, msg := range msgs {
if mode != runTxModeDeliver && mode != runTxModeSimulate {
break
}
handler := app.msgServiceRouter.Handler(msg)
if handler == nil {
return nil, sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "can't route message %+v", msg)
}
/ ADR 031 request type routing
msgResult, err := handler(ctx, msg)
if err != nil {
return nil, sdkerrors.Wrapf(err, "failed to execute message; message index: %d", i)
}
/ create message events
msgEvents := createEvents(msgResult.GetEvents(), msg)
/ append message events, data and logs
/
/ Note: Each message result's data must be length-prefixed in order to
/ separate each result.
events = events.AppendEvents(msgEvents)
/ Each individual sdk.Result that went through the MsgServiceRouter
/ (which should represent 99% of the Msgs now, since everyone should
/ be using protobuf Msgs)
has exactly one Msg response, set inside
/ `WrapServiceResult`. We take that Msg response, and aggregate it
/ into an array.
if len(msgResult.MsgResponses) > 0 {
msgResponse := msgResult.MsgResponses[0]
if msgResponse == nil {
return nil, sdkerrors.ErrLogic.Wrapf("got nil Msg response at index %d for msg %s", i, sdk.MsgTypeURL(msg))
}
msgResponses = append(msgResponses, msgResponse)
}
msgLogs = append(msgLogs, sdk.NewABCIMessageLog(uint32(i), msgResult.Log, msgEvents))
}
data, err := makeABCIData(msgResponses)
if err != nil {
return nil, sdkerrors.Wrap(err, "failed to marshal tx data")
}
return &sdk.Result{
Data: data,
Log: strings.TrimSpace(msgLogs.String()),
Events: events.ToABCIEvents(),
MsgResponses: msgResponses,
}, nil
}
/ makeABCIData generates the Data field to be sent to ABCI Check/DeliverTx.
func makeABCIData(msgResponses []*codectypes.Any) ([]byte, error) {
return proto.Marshal(&sdk.TxMsgData{
MsgResponses: msgResponses
})
}
func createEvents(events sdk.Events, msg sdk.Msg)
sdk.Events {
eventMsgName := sdk.MsgTypeURL(msg)
msgEvent := sdk.NewEvent(sdk.EventTypeMessage, sdk.NewAttribute(sdk.AttributeKeyAction, eventMsgName))
/ we set the signer attribute as the sender
if len(msg.GetSigners()) > 0 && !msg.GetSigners()[0].Empty() {
msgEvent = msgEvent.AppendAttributes(sdk.NewAttribute(sdk.AttributeKeySender, msg.GetSigners()[0].String()))
}
/ verify that events have no module attribute set
if _, found := events.GetAttributes(sdk.AttributeKeyModule); !found {
/ here we assume that routes module name is the second element of the route
/ e.g. "cosmos.bank.v1beta1.MsgSend" => "bank"
moduleName := strings.Split(eventMsgName, ".")
if len(moduleName) > 1 {
msgEvent = msgEvent.AppendAttributes(sdk.NewAttribute(sdk.AttributeKeyModule, moduleName[1]))
}
}
return sdk.Events{
msgEvent
}.AppendEvents(events)
}
/ DefaultPrepareProposal returns the default implementation for processing an
/ ABCI proposal. The application's mempool is enumerated and all valid
/ transactions are added to the proposal. Transactions are valid if they:
/
/ 1)
Successfully encode to bytes.
/ 2)
Are valid (i.e. pass runTx, AnteHandler only).
/
/ Enumeration is halted once RequestPrepareProposal.MaxBytes of transactions is
/ reached or the mempool is exhausted.
/
/ Note:
/
/ - Step (2)
is identical to the validation step performed in
/ DefaultProcessProposal. It is very important that the same validation logic
/ is used in both steps, and applications must ensure that this is the case in
/ non-default handlers.
/
/ - If no mempool is set or if the mempool is a no-op mempool, the transactions
/ requested from Tendermint will simply be returned, which, by default, are in
/ FIFO order.
func (app *BaseApp)
DefaultPrepareProposal()
sdk.PrepareProposalHandler {
return func(ctx sdk.Context, req abci.RequestPrepareProposal)
abci.ResponsePrepareProposal {
/ If the mempool is nil or a no-op mempool, we simply return the transactions
/ requested from Tendermint, which, by default, should be in FIFO order.
_, isNoOp := app.mempool.(mempool.NoOpMempool)
if app.mempool == nil || isNoOp {
return abci.ResponsePrepareProposal{
Txs: req.Txs
}
}
var (
txsBytes [][]byte
byteCount int64
)
iterator := app.mempool.Select(ctx, req.Txs)
for iterator != nil {
memTx := iterator.Tx()
bz, err := app.txEncoder(memTx)
if err != nil {
panic(err)
}
txSize := int64(len(bz))
/ NOTE: Since runTx was already executed in CheckTx, which calls
/ mempool.Insert, ideally everything in the pool should be valid. But
/ some mempool implementations may insert invalid txs, so we check again.
_, _, _, _, err = app.runTx(runTxPrepareProposal, bz)
if err != nil {
err := app.mempool.Remove(memTx)
if err != nil && !errors.Is(err, mempool.ErrTxNotFound) {
panic(err)
}
iterator = iterator.Next()
continue
}
else if byteCount += txSize; byteCount <= req.MaxTxBytes {
txsBytes = append(txsBytes, bz)
}
else {
break
}
iterator = iterator.Next()
}
return abci.ResponsePrepareProposal{
Txs: txsBytes
}
}
}
/ DefaultProcessProposal returns the default implementation for processing an ABCI proposal.
/ Every transaction in the proposal must pass 2 conditions:
/
/ 1. The transaction bytes must decode to a valid transaction.
/ 2. The transaction must be valid (i.e. pass runTx, AnteHandler only)
/
/ If any transaction fails to pass either condition, the proposal is rejected. Note that step (2)
is identical to the
/ validation step performed in DefaultPrepareProposal. It is very important that the same validation logic is used
/ in both steps, and applications must ensure that this is the case in non-default handlers.
func (app *BaseApp)
DefaultProcessProposal()
sdk.ProcessProposalHandler {
return func(ctx sdk.Context, req abci.RequestProcessProposal)
abci.ResponseProcessProposal {
for _, txBytes := range req.Txs {
_, err := app.txDecoder(txBytes)
if err != nil {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_REJECT
}
}
_, _, _, _, err = app.runTx(runTxProcessProposal, txBytes)
if err != nil {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_REJECT
}
}
}
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_ACCEPT
}
}
}
/ NoOpPrepareProposal defines a no-op PrepareProposal handler. It will always
/ return the transactions sent by the client's request.
func NoOpPrepareProposal()
sdk.PrepareProposalHandler {
return func(_ sdk.Context, req abci.RequestPrepareProposal)
abci.ResponsePrepareProposal {
return abci.ResponsePrepareProposal{
Txs: req.Txs
}
}
}
/ NoOpProcessProposal defines a no-op ProcessProposal Handler. It will always
/ return ACCEPT.
func NoOpProcessProposal()
sdk.ProcessProposalHandler {
return func(_ sdk.Context, _ abci.RequestProcessProposal)
abci.ResponseProcessProposal {
return abci.ResponseProcessProposal{
Status: abci.ResponseProcessProposal_ACCEPT
}
}
}
PrepareProposal this implementation is the default and can be modified by the application developer in app.go:
Copy
Ask AI
processOpt := func(app *baseapp.BaseApp) {
abciPropHandler := baseapp.NewDefaultProposalHandler(mempool, app)
app.SetProcessProposal(abciPropHandler.ProcessProposalHandler())
}
baseAppOptions = append(baseAppOptions, processOpt)
Mempool
Now that we have walked through thePrepareProposal & ProcessProposal, we can move on to walking through the mempool.
There are countless designs that an application developer can write for a mempool, the SDK opted to provide only simple mempool implementations.
Namely, the SDK provides the following mempools:
The default SDK is a No-op Mempool, but it can be replaced by the application developer in app.go:
Copy
Ask AI
nonceMempool := mempool.NewSenderNonceMempool()
mempoolOpt := baseapp.SetMempool(nonceMempool)
baseAppOptions = append(baseAppOptions, mempoolOpt)
No-op Mempool
A no-op mempool is a mempool where transactions are completely discarded and ignored when BaseApp interacts with the mempool. When this mempool is used, it assumed that an application will rely on CometBFT’s transaction ordering defined inRequestPrepareProposal,
which is FIFO-ordered by default.
Sender Nonce Mempool
The nonce mempool is a mempool that keeps transactions from an sorted by nonce in order to avoid the issues with nonces. It works by storing the transaction in a list sorted by the transaction nonce. When the proposer asks for transactions to be included in a block it randomly selects a sender and gets the first transaction in the list. It repeats this until the mempool is empty or the block is full. It is configurable with the following parameters:MaxTxs
It is an integer value that sets the mempool in one of three modes, bounded, unbounded, or disabled.- negative: Disabled, mempool does not insert new transaction and return early.
- zero: Unbounded mempool has no transaction limit and will never fail with
ErrMempoolTxMaxCapacity. - positive: Bounded, it fails with
ErrMempoolTxMaxCapacitywhenmaxTxvalue is the same asCountTx()
Seed
Set the seed for the random number generator used to select transactions from the mempool.Priority Nonce Mempool
The priority nonce mempool is a mempool implementation that stores txs in a partially ordered set by 2 dimensions:- priority
- sender-nonce (sequence number)
MaxTxs
It is an integer value that sets the mempool in one of three modes, bounded, unbounded, or disabled.- negative: Disabled, mempool does not insert new transaction and return early.
- zero: Unbounded mempool has no transaction limit and will never fail with
ErrMempoolTxMaxCapacity. - positive: Bounded, it fails with
ErrMempoolTxMaxCapacitywhenmaxTxvalue is the same asCountTx()
Callback
The priority nonce mempool provides mempool options allowing the application sets callback(s).- OnRead: Set a callback to be called when a transaction is read from the mempool.
- TxReplacement: Sets a callback to be called when duplicated transaction nonce detected during mempool insert. Application can define a transaction replacement rule based on tx priority or certain transaction fields.