refactor(stf): remove RunWithCtx (#21739)

This commit is contained in:
Matt Kocubinski 2024-09-19 16:39:47 -05:00 committed by GitHub
parent 98eb0b7cd6
commit d6364b8956
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 218 additions and 85 deletions

View File

@ -35,7 +35,7 @@ func (i *Info) Bytes() ([]byte, error) {
// Encode Hash
if len(i.Hash) != hashSize {
return nil, errors.New("invalid hash size")
return nil, errors.New("invalid Hash size")
}
buf = append(buf, i.Hash...)
@ -47,7 +47,7 @@ func (i *Info) Bytes() ([]byte, error) {
// Encode AppHash
if len(i.AppHash) != hashSize {
return nil, errors.New("invalid hash size")
return nil, errors.New("invalid AppHash size")
}
buf = append(buf, i.AppHash...)

View File

@ -10,6 +10,11 @@ type KVStoreService interface {
OpenKVStore(context.Context) KVStore
}
// KVStoreServiceFactory is a function that creates a new KVStoreService.
// It can be used to override the default KVStoreService bindings for cases
// where an application must supply a custom stateful backend.
type KVStoreServiceFactory func([]byte) KVStoreService
// MemoryStoreService represents a unique, non-forgeable handle to a memory-backed
// KVStore. It should be provided as a module-scoped dependency by the runtime
// module being used to build the app.

View File

@ -3,6 +3,7 @@ package runtime
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"path/filepath"
@ -12,6 +13,7 @@ import (
"cosmossdk.io/core/server"
"cosmossdk.io/core/store"
"cosmossdk.io/core/transaction"
"cosmossdk.io/runtime/v2/services"
"cosmossdk.io/server/v2/appmanager"
"cosmossdk.io/server/v2/stf"
"cosmossdk.io/server/v2/stf/branch"
@ -157,23 +159,51 @@ func (a *AppBuilder[T]) Build(opts ...AppBuilderOption[T]) (*App[T], error) {
ValidateTxGasLimit: a.app.config.GasConfig.ValidateTxGasLimit,
QueryGasLimit: a.app.config.GasConfig.QueryGasLimit,
SimulationGasLimit: a.app.config.GasConfig.SimulationGasLimit,
InitGenesis: func(ctx context.Context, src io.Reader, txHandler func(json.RawMessage) error) error {
InitGenesis: func(
ctx context.Context,
src io.Reader,
txHandler func(json.RawMessage) error,
) (store.WriterMap, error) {
// this implementation assumes that the state is a JSON object
bz, err := io.ReadAll(src)
if err != nil {
return fmt.Errorf("failed to read import state: %w", err)
return nil, fmt.Errorf("failed to read import state: %w", err)
}
var genesisState map[string]json.RawMessage
if err = json.Unmarshal(bz, &genesisState); err != nil {
return err
var genesisJSON map[string]json.RawMessage
if err = json.Unmarshal(bz, &genesisJSON); err != nil {
return nil, err
}
if err = a.app.moduleManager.InitGenesisJSON(ctx, genesisState, txHandler); err != nil {
return fmt.Errorf("failed to init genesis: %w", err)
v, zeroState, err := a.app.db.StateLatest()
if err != nil {
return nil, fmt.Errorf("unable to get latest state: %w", err)
}
return nil
if v != 0 { // TODO: genesis state may be > 0, we need to set version on store
return nil, errors.New("cannot init genesis on non-zero state")
}
genesisCtx := services.NewGenesisContext(a.branch(zeroState))
genesisState, err := genesisCtx.Run(ctx, func(ctx context.Context) error {
err = a.app.moduleManager.InitGenesisJSON(ctx, genesisJSON, txHandler)
if err != nil {
return fmt.Errorf("failed to init genesis: %w", err)
}
return nil
})
return genesisState, err
},
ExportGenesis: func(ctx context.Context, version uint64) ([]byte, error) {
genesisJson, err := a.app.moduleManager.ExportGenesisForModules(ctx)
_, state, err := a.app.db.StateLatest()
if err != nil {
return nil, fmt.Errorf("unable to get latest state: %w", err)
}
genesisCtx := services.NewGenesisContext(a.branch(state))
var genesisJson map[string]json.RawMessage
_, err = genesisCtx.Run(ctx, func(ctx context.Context) error {
genesisJson, err = a.app.moduleManager.ExportGenesisForModules(ctx)
return err
})
if err != nil {
return nil, fmt.Errorf("failed to export genesis: %w", err)
}

View File

@ -5,6 +5,7 @@ go 1.23
// server v2 integration
replace (
cosmossdk.io/api => ../../api
cosmossdk.io/core => ../../core
cosmossdk.io/core/testing => ../../core/testing
cosmossdk.io/server/v2/appmanager => ../../server/v2/appmanager
cosmossdk.io/server/v2/stf => ../../server/v2/stf

View File

@ -2,8 +2,6 @@ buf.build/gen/go/cometbft/cometbft/protocolbuffers/go v1.34.2-20240701160653-fed
buf.build/gen/go/cometbft/cometbft/protocolbuffers/go v1.34.2-20240701160653-fedbb9acfd2f.2/go.mod h1:1+3gJj2NvZ1mTLAtHu+lMhOjGgQPiCKCeo+9MBww0Eo=
buf.build/gen/go/cosmos/gogo-proto/protocolbuffers/go v1.34.2-20240130113600-88ef6483f90f.2 h1:b7EEYTUHmWSBEyISHlHvXbJPqtKiHRuUignL1tsHnNQ=
buf.build/gen/go/cosmos/gogo-proto/protocolbuffers/go v1.34.2-20240130113600-88ef6483f90f.2/go.mod h1:HqcXMSa5qnNuakaMUo+hWhF51mKbcrZxGl9Vp5EeJXc=
cosmossdk.io/core v1.0.0-alpha.3 h1:pnxaYAas7llXgVz1lM7X6De74nWrhNKnB3yMKe4OUUA=
cosmossdk.io/core v1.0.0-alpha.3/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
cosmossdk.io/depinject v1.0.0 h1:dQaTu6+O6askNXO06+jyeUAnF2/ssKwrrszP9t5q050=
cosmossdk.io/depinject v1.0.0/go.mod h1:zxK/h3HgHoA/eJVtiSsoaRaRA2D5U4cJ5thIG4ssbB8=
cosmossdk.io/errors/v2 v2.0.0-20240731132947-df72853b3ca5 h1:IQNdY2kB+k+1OM2DvqFG1+UgeU1JzZrWtwuWzI3ZfwA=

View File

@ -16,6 +16,7 @@ import (
reflectionv1 "cosmossdk.io/api/cosmos/reflection/v1"
appmodulev2 "cosmossdk.io/core/appmodule/v2"
"cosmossdk.io/core/comet"
"cosmossdk.io/core/header"
"cosmossdk.io/core/registry"
"cosmossdk.io/core/server"
"cosmossdk.io/core/store"
@ -96,7 +97,6 @@ func init() {
ProvideAppBuilder[transaction.Tx],
ProvideEnvironment[transaction.Tx],
ProvideModuleManager[transaction.Tx],
ProvideCometService,
),
appconfig.Invoke(SetupAppBuilder),
)
@ -176,6 +176,8 @@ func ProvideEnvironment[T transaction.Tx](
config *runtimev2.Module,
key depinject.ModuleKey,
appBuilder *AppBuilder[T],
kvFactory store.KVStoreServiceFactory,
headerService header.Service,
) (
appmodulev2.Environment,
store.KVStoreService,
@ -197,7 +199,7 @@ func ProvideEnvironment[T transaction.Tx](
}
registerStoreKey(appBuilder, kvStoreKey)
kvService = stf.NewKVStoreService([]byte(kvStoreKey))
kvService = kvFactory([]byte(kvStoreKey))
memStoreKey := fmt.Sprintf("memory:%s", key.Name())
registerStoreKey(appBuilder, memStoreKey)
@ -209,7 +211,7 @@ func ProvideEnvironment[T transaction.Tx](
BranchService: stf.BranchService{},
EventService: stf.NewEventService(),
GasService: stf.NewGasMeterService(),
HeaderService: stf.HeaderService{},
HeaderService: headerService,
QueryRouterService: stf.NewQueryRouterService(),
MsgRouterService: stf.NewMsgRouterService([]byte(key.Name())),
TransactionService: services.NewContextAwareTransactionService(),
@ -220,8 +222,8 @@ func ProvideEnvironment[T transaction.Tx](
return env, kvService, memKvService
}
func registerStoreKey[T transaction.Tx](wrapper *AppBuilder[T], key string) {
wrapper.app.storeKeys = append(wrapper.app.storeKeys, key)
func registerStoreKey[T transaction.Tx](builder *AppBuilder[T], key string) {
builder.app.storeKeys = append(builder.app.storeKeys, key)
}
func storeKeyOverride(config *runtimev2.Module, moduleName string) *runtimev2.StoreKeyConfig {
@ -234,6 +236,28 @@ func storeKeyOverride(config *runtimev2.Module, moduleName string) *runtimev2.St
return nil
}
func ProvideCometService() comet.Service {
return &services.ContextAwareCometInfoService{}
// DefaultServiceBindings provides default services for the following service interfaces:
// - store.KVStoreServiceFactory
// - header.Service
// - comet.Service
//
// They are all required. For most use cases these default services bindings should be sufficient.
// Power users (or tests) may wish to provide their own services bindings, in which case they must
// supply implementations for each of the above interfaces.
func DefaultServiceBindings() depinject.Config {
var (
kvServiceFactory store.KVStoreServiceFactory = func(actor []byte) store.KVStoreService {
return services.NewGenesisKVService(
actor,
stf.NewKVStoreService(actor),
)
}
headerService header.Service = services.NewGenesisHeaderService(stf.HeaderService{})
cometService comet.Service = &services.ContextAwareCometInfoService{}
)
return depinject.Supply(
kvServiceFactory,
headerService,
cometService,
)
}

View File

@ -0,0 +1,107 @@
package services
import (
"context"
"fmt"
"cosmossdk.io/core/header"
"cosmossdk.io/core/store"
)
var (
_ store.KVStoreService = (*GenesisKVStoreService)(nil)
_ header.Service = (*GenesisHeaderService)(nil)
)
type genesisContextKeyType struct{}
var genesisContextKey = genesisContextKeyType{}
// genesisContext is a context that is used during genesis initialization.
// it backs the store.KVStoreService and header.Service interface implementations
// defined in this file.
type genesisContext struct {
state store.WriterMap
}
// NewGenesisContext creates a new genesis context.
func NewGenesisContext(state store.WriterMap) genesisContext {
return genesisContext{
state: state,
}
}
// Run runs the provided function within the genesis context and returns an
// updated store.WriterMap containing the state modifications made during InitGenesis.
func (g *genesisContext) Run(
ctx context.Context,
fn func(ctx context.Context) error,
) (store.WriterMap, error) {
ctx = context.WithValue(ctx, genesisContextKey, g)
err := fn(ctx)
if err != nil {
return nil, err
}
return g.state, nil
}
// GenesisKVStoreService is a store.KVStoreService implementation that is used during
// genesis initialization. It wraps an inner execution context store.KVStoreService.
type GenesisKVStoreService struct {
actor []byte
executionService store.KVStoreService
}
// NewGenesisKVService creates a new GenesisKVStoreService.
// - actor is the module store key.
// - executionService is the store.KVStoreService to use when the genesis context is not active.
func NewGenesisKVService(
actor []byte,
executionService store.KVStoreService,
) *GenesisKVStoreService {
return &GenesisKVStoreService{
actor: actor,
executionService: executionService,
}
}
// OpenKVStore implements store.KVStoreService.
func (g *GenesisKVStoreService) OpenKVStore(ctx context.Context) store.KVStore {
v := ctx.Value(genesisContextKey)
if v == nil {
return g.executionService.OpenKVStore(ctx)
}
genCtx, ok := v.(*genesisContext)
if !ok {
panic(fmt.Errorf("unexpected genesis context type: %T", v))
}
state, err := genCtx.state.GetWriter(g.actor)
if err != nil {
panic(err)
}
return state
}
// GenesisHeaderService is a header.Service implementation that is used during
// genesis initialization. It wraps an inner execution context header.Service.
type GenesisHeaderService struct {
executionService header.Service
}
// HeaderInfo implements header.Service.
// During genesis initialization, it returns an empty header.Info.
func (g *GenesisHeaderService) HeaderInfo(ctx context.Context) header.Info {
v := ctx.Value(genesisContextKey)
if v == nil {
return g.executionService.HeaderInfo(ctx)
}
return header.Info{}
}
// NewGenesisHeaderService creates a new GenesisHeaderService.
// - executionService is the header.Service to use when the genesis context is not active.
func NewGenesisHeaderService(executionService header.Service) *GenesisHeaderService {
return &GenesisHeaderService{
executionService: executionService,
}
}

View File

@ -43,25 +43,19 @@ func (a AppManager[T]) InitGenesis(
initGenesisJSON []byte,
txDecoder transaction.Codec[T],
) (*server.BlockResponse, corestore.WriterMap, error) {
v, zeroState, err := a.db.StateLatest()
if err != nil {
return nil, nil, fmt.Errorf("unable to get latest state: %w", err)
}
if v != 0 { // TODO: genesis state may be > 0, we need to set version on store
return nil, nil, errors.New("cannot init genesis on non-zero state")
}
var genTxs []T
genesisState, err := a.stf.RunWithCtx(ctx, zeroState, func(ctx context.Context) error {
return a.initGenesis(ctx, bytes.NewBuffer(initGenesisJSON), func(jsonTx json.RawMessage) error {
genesisState, err := a.initGenesis(
ctx,
bytes.NewBuffer(initGenesisJSON),
func(jsonTx json.RawMessage) error {
genTx, err := txDecoder.DecodeJSON(jsonTx)
if err != nil {
return fmt.Errorf("failed to decode genesis transaction: %w", err)
}
genTxs = append(genTxs, genTx)
return nil
})
})
},
)
if err != nil {
return nil, nil, fmt.Errorf("failed to import genesis state: %w", err)
}
@ -89,29 +83,11 @@ func (a AppManager[T]) InitGenesis(
// ExportGenesis exports the genesis state of the application.
func (a AppManager[T]) ExportGenesis(ctx context.Context, version uint64) ([]byte, error) {
zeroState, err := a.db.StateAt(version)
if err != nil {
return nil, fmt.Errorf("unable to get latest state: %w", err)
if a.exportGenesis == nil {
return nil, errors.New("export genesis function not set")
}
bz := make([]byte, 0)
_, err = a.stf.RunWithCtx(ctx, zeroState, func(ctx context.Context) error {
if a.exportGenesis == nil {
return errors.New("export genesis function not set")
}
bz, err = a.exportGenesis(ctx, version)
if err != nil {
return fmt.Errorf("failed to export genesis state: %w", err)
}
return nil
})
if err != nil {
return nil, fmt.Errorf("failed to export genesis state: %w", err)
}
return bz, nil
return a.exportGenesis(ctx, version)
}
func (a AppManager[T]) DeliverBlock(
@ -180,10 +156,6 @@ func (a AppManager[T]) Query(ctx context.Context, version uint64, request transa
// QueryWithState executes a query with the provided state. This allows to process a query
// independently of the db state. For example, it can be used to process a query with temporary
// and uncommitted state
func (a AppManager[T]) QueryWithState(
ctx context.Context,
state corestore.ReaderMap,
request transaction.Msg,
) (transaction.Msg, error) {
func (a AppManager[T]) QueryWithState(ctx context.Context, state corestore.ReaderMap, request transaction.Msg) (transaction.Msg, error) {
return a.stf.Query(ctx, state, a.config.QueryGasLimit, request)
}

View File

@ -4,11 +4,25 @@ import (
"context"
"encoding/json"
"io"
"cosmossdk.io/core/store"
)
type (
// ExportGenesis is a function type that represents the export of the genesis state.
ExportGenesis func(ctx context.Context, version uint64) ([]byte, error)
// InitGenesis is a function type that represents the initialization of the genesis state.
InitGenesis func(ctx context.Context, src io.Reader, txHandler func(json.RawMessage) error) error
// InitGenesis is a function that will run at application genesis, it will be called with
// the following arguments:
// - ctx: the context of the genesis operation
// - src: the source containing the raw genesis state
// - txHandler: a function capable of decoding a json tx, will be run for each genesis
// transaction
//
// It must return a map of the dirty state after the genesis operation.
InitGenesis func(
ctx context.Context,
src io.Reader,
txHandler func(json.RawMessage) error,
) (store.WriterMap, error)
)

View File

@ -40,12 +40,4 @@ type StateTransitionFunction[T transaction.Tx] interface {
gasLimit uint64,
req transaction.Msg,
) (transaction.Msg, error)
// RunWithCtx executes the provided closure within a context.
// TODO: remove
RunWithCtx(
ctx context.Context,
state store.ReaderMap,
closure func(ctx context.Context) error,
) (store.WriterMap, error)
}

View File

@ -678,8 +678,10 @@ func setUpConsensus(t *testing.T, gasLimit uint64, mempool mempool.Mempool[mock.
ValidateTxGasLimit: gasLimit,
QueryGasLimit: gasLimit,
SimulationGasLimit: gasLimit,
InitGenesis: func(ctx context.Context, src io.Reader, txHandler func(json.RawMessage) error) error {
return nil
InitGenesis: func(ctx context.Context, src io.Reader, txHandler func(json.RawMessage) error) (store.WriterMap, error) {
_, st, err := mockStore.StateLatest()
require.NoError(t, err)
return branch.DefaultNewWriterMap(st), nil
},
}

View File

@ -448,19 +448,6 @@ func (s STF[T]) Query(
return s.queryRouter.Invoke(queryCtx, req)
}
// RunWithCtx is made to support genesis, if genesis was just the execution of messages instead
// of being something custom then we would not need this. PLEASE DO NOT USE.
// TODO: Remove
func (s STF[T]) RunWithCtx(
ctx context.Context,
state store.ReaderMap,
closure func(ctx context.Context) error,
) (store.WriterMap, error) {
branchedState := s.branchFn(state)
stfCtx := s.makeContext(ctx, nil, branchedState, internal.ExecModeFinalize)
return branchedState, closure(stfCtx)
}
// clone clones STF.
func (s STF[T]) clone() STF[T] {
return STF[T]{

View File

@ -69,6 +69,7 @@ func NewSimApp[T transaction.Tx](
// merge the AppConfig and other configuration in one config
appConfig = depinject.Configs(
AppConfig(),
runtime.DefaultServiceBindings(),
depinject.Supply(
logger,
viper,

View File

@ -288,6 +288,7 @@ replace (
// server v2 integration
replace (
cosmossdk.io/api => ../../api
cosmossdk.io/core => ../../core
cosmossdk.io/core/testing => ../../core/testing
cosmossdk.io/runtime/v2 => ../../runtime/v2
cosmossdk.io/server/v2 => ../../server/v2

View File

@ -192,8 +192,6 @@ cloud.google.com/go/webrisk v1.4.0/go.mod h1:Hn8X6Zr+ziE2aNd8SliSDWpEnSS1u4R9+xX
cloud.google.com/go/webrisk v1.5.0/go.mod h1:iPG6fr52Tv7sGk0H6qUFzmL3HHZev1htXuWDEEsqMTg=
cloud.google.com/go/workflows v1.6.0/go.mod h1:6t9F5h/unJz41YqfBmqSASJSXccBLtD1Vwf+KmJENM0=
cloud.google.com/go/workflows v1.7.0/go.mod h1:JhSrZuVZWuiDfKEFxU0/F1PQjmpnpcoISEXH2bcHC3M=
cosmossdk.io/core v1.0.0-alpha.3 h1:pnxaYAas7llXgVz1lM7X6De74nWrhNKnB3yMKe4OUUA=
cosmossdk.io/core v1.0.0-alpha.3/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
cosmossdk.io/depinject v1.0.0 h1:dQaTu6+O6askNXO06+jyeUAnF2/ssKwrrszP9t5q050=
cosmossdk.io/depinject v1.0.0/go.mod h1:zxK/h3HgHoA/eJVtiSsoaRaRA2D5U4cJ5thIG4ssbB8=
cosmossdk.io/errors v1.0.1 h1:bzu+Kcr0kS/1DuPBtUFdWjzLqyUuCiyHjyJB6srBV/0=

View File

@ -37,6 +37,7 @@ func NewRootCmd[T transaction.Tx]() *cobra.Command {
if err := depinject.Inject(
depinject.Configs(
simapp.AppConfig(),
runtime.DefaultServiceBindings(),
depinject.Supply(log.NewNopLogger()),
depinject.Provide(
codec.ProvideInterfaceRegistry,