Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor(server/v2/cometbft): Handle non-module service queries #22803

Merged
merged 17 commits into from
Dec 11, 2024
176 changes: 144 additions & 32 deletions server/v2/cometbft/abci.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"cosmossdk.io/core/event"
"cosmossdk.io/core/server"
"cosmossdk.io/core/store"
"cosmossdk.io/core/transaction"
errorsmod "cosmossdk.io/errors/v2"
"cosmossdk.io/log"
"cosmossdk.io/schema/appdata"
Expand All @@ -35,10 +34,15 @@ import (
"cosmossdk.io/store/v2/snapshots"
consensustypes "cosmossdk.io/x/consensus/types"

addresscodec "cosmossdk.io/core/address"
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
coreserver "cosmossdk.io/core/server"
"cosmossdk.io/core/transaction"
rpchttp "github.com/cometbft/cometbft/rpc/client/http"
"github.com/cosmos/cosmos-sdk/client"
"github.com/cosmos/cosmos-sdk/client/grpc/cmtservice"
"github.com/cosmos/cosmos-sdk/codec"
codectypes "github.com/cosmos/cosmos-sdk/codec/types"
sdk "github.com/cosmos/cosmos-sdk/types"
txtypes "github.com/cosmos/cosmos-sdk/types/tx"
"github.com/cosmos/cosmos-sdk/std"
authtx "github.com/cosmos/cosmos-sdk/x/auth/tx"
)

const (
Expand Down Expand Up @@ -86,8 +90,10 @@ type consensus[T transaction.Tx] struct {
addrPeerFilter types.PeerFilter // filter peers by address and port
idPeerFilter types.PeerFilter // filter peers by node ID

queryHandlersMap map[string]appmodulev2.Handler
getProtoRegistry func() (*protoregistry.Files, error)
queryHandlersMap map[string]appmodulev2.Handler
getProtoRegistry func() (*protoregistry.Files, error)
consensusAddressCodec addresscodec.Codec
cfgMap coreserver.ConfigMap
}

// CheckTx implements types.Application.
Expand Down Expand Up @@ -184,6 +190,15 @@ func (c *consensus[T]) Query(ctx context.Context, req *abciproto.QueryRequest) (
return resp, err
}

// when a client did not provide a query height, manually inject the latest
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
if req.Height == 0 {
lastestVersion, err := c.store.GetLatestVersion()
if err != nil {
return nil, err
}
req.Height = int64(lastestVersion)
hieuvubk marked this conversation as resolved.
Show resolved Hide resolved
}

// this error most probably means that we can't handle it with a proto message, so
// it must be an app/p2p/store query
path := splitABCIQueryPath(req.Path)
Expand Down Expand Up @@ -238,44 +253,117 @@ func (c *consensus[T]) maybeRunGRPCQuery(ctx context.Context, req *abci.QueryReq
handlerFullName = string(md.Input().FullName())
}

// special case for simulation as it is an external gRPC registered on the grpc server component
// special case for non-module services as they are external gRPC registered on the grpc server component
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
// and not on the app itself, so it won't pass the router afterwards.
if req.Path == "/cosmos.tx.v1beta1.Service/Simulate" {
simulateRequest := &txtypes.SimulateRequest{}
err = gogoproto.Unmarshal(req.Data, simulateRequest)
if err != nil {
return nil, true, fmt.Errorf("unable to decode gRPC request with path %s from ABCI.Query: %w", req.Path, err)

// Handle comet service
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
if strings.Contains(req.Path, "/cosmos.base.tendermint.v1beta1.Service") {
rpcClient, _ := rpchttp.New(c.cfg.AppTomlConfig.Address)
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
cometQServer := cmtservice.NewQueryServer(rpcClient, c.Query, c.consensusAddressCodec)
paths := strings.Split(req.Path, "/")

var resp transaction.Msg
var err error
switch paths[2] {
hieuvubk marked this conversation as resolved.
Show resolved Hide resolved
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
case "GetNodeInfo":
resp, err = handleCometService(ctx, req, cometQServer.GetNodeInfo)
case "GetSyncing":
resp, err = handleCometService(ctx, req, cometQServer.GetSyncing)
case "GetLatestBlock":
resp, err = handleCometService(ctx, req, cometQServer.GetLatestBlock)
case "GetBlockByHeight":
resp, err = handleCometService(ctx, req, cometQServer.GetBlockByHeight)
case "GetLatestValidatorSet":
resp, err = handleCometService(ctx, req, cometQServer.GetLatestValidatorSet)
case "GetValidatorSetByHeight":
resp, err = handleCometService(ctx, req, cometQServer.GetValidatorSetByHeight)
case "ABCIQuery":
resp, err = handleCometService(ctx, req, cometQServer.ABCIQuery)
}

tx, err := c.txCodec.Decode(simulateRequest.TxBytes)
if err != nil {
return nil, true, fmt.Errorf("failed to decode tx: %w", err)
return nil, true, err
}

res, err := queryResponse(resp, req.Height)
return res, true, err
}

// Handle node service
if strings.Contains(req.Path, "/cosmos.base.node.v1beta1.Service") {
nodeQService := nodeServer[T]{c.cfgMap, c.cfg.AppTomlConfig, c}
paths := strings.Split(req.Path, "/")
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved

var resp transaction.Msg
var err error
switch paths[2] {
case "Config":
resp, err = handleCometService(ctx, req, nodeQService.Config)
case "Status":
resp, err = handleCometService(ctx, req, nodeQService.Status)
}

txResult, _, err := c.app.Simulate(ctx, tx)
if err != nil {
return nil, true, fmt.Errorf("failed with gas used: '%d': %w", txResult.GasUsed, err)
return nil, true, err
}

msgResponses := make([]*codectypes.Any, 0, len(txResult.Resp))
// pack the messages into Any
for _, msg := range txResult.Resp {
anyMsg, err := codectypes.NewAnyWithValue(msg)
if err != nil {
return nil, true, fmt.Errorf("failed to pack message response: %w", err)
}
res, err := queryResponse(resp, req.Height)
return res, true, err
}

msgResponses = append(msgResponses, anyMsg)
// Handle tx service
if strings.Contains(req.Path, "/cosmos.tx.v1beta1.Service") {
// init simple client context
amino := codec.NewLegacyAmino()
std.RegisterLegacyAminoCodec(amino)
txConfig := authtx.NewTxConfig(
c.appCodec,
c.appCodec.InterfaceRegistry().SigningContext().AddressCodec(),
c.appCodec.InterfaceRegistry().SigningContext().ValidatorAddressCodec(),
authtx.DefaultSignModes,
)
rpcClient, _ := client.NewClientFromNode(c.cfg.AppTomlConfig.Address)

clientCtx := client.Context{}.
WithLegacyAmino(amino).
WithCodec(c.appCodec).
WithTxConfig(txConfig).
WithNodeURI(c.cfg.AppTomlConfig.Address).
WithClient(rpcClient)

txService := txServer[T]{
clientCtx: clientCtx,
txCodec: c.txCodec,
app: c.app,
consensus: c,
}
paths := strings.Split(req.Path, "/")

var resp transaction.Msg
var err error
switch paths[2] {
case "Simulate":
resp, err = handleCometService(ctx, req, txService.Simulate)
case "GetTx":
resp, err = handleCometService(ctx, req, txService.GetTx)
case "BroadcastTx":
return nil, true, errors.New("can't route a broadcast tx message")
case "GetTxsEvent":
resp, err = handleCometService(ctx, req, txService.GetTxsEvent)
case "GetBlockWithTxs":
resp, err = handleCometService(ctx, req, txService.GetBlockWithTxs)
case "TxDecode":
resp, err = handleCometService(ctx, req, txService.TxDecode)
case "TxEncode":
resp, err = handleCometService(ctx, req, txService.TxEncode)
case "TxEncodeAmino":
resp, err = handleCometService(ctx, req, txService.TxEncodeAmino)
case "TxDecodeAmino":
resp, err = handleCometService(ctx, req, txService.Simulate)
}

resp := &txtypes.SimulateResponse{
GasInfo: &sdk.GasInfo{
GasUsed: txResult.GasUsed,
GasWanted: txResult.GasWanted,
},
Result: &sdk.Result{
MsgResponses: msgResponses,
},
if err != nil {
return nil, true, err
}

res, err := queryResponse(resp, req.Height)
Expand Down Expand Up @@ -303,6 +391,30 @@ func (c *consensus[T]) maybeRunGRPCQuery(ctx context.Context, req *abci.QueryReq
return resp, true, err
}

func handleCometService[T any, PT interface {
*T
gogoproto.Message
},
U any, UT interface {
*U
gogoproto.Message
}](
ctx context.Context,
rawReq *abciproto.QueryRequest,
handler func(ctx context.Context, msg PT) (UT, error),
) (transaction.Msg, error) {
req := PT(new(T))
err := gogoproto.Unmarshal(rawReq.Data, req)
if err != nil {
return nil, err
}
typedResp, err := handler(ctx, req)
if err != nil {
return nil, err
}
return typedResp, nil
}

// InitChain implements types.Application.
func (c *consensus[T]) InitChain(ctx context.Context, req *abciproto.InitChainRequest) (*abciproto.InitChainResponse, error) {
c.logger.Info("InitChain", "initialHeight", req.InitialHeight, "chainID", req.ChainId)
Expand Down
Loading
Loading