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.

Sign up for GitHub

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

Jump to bottom

refactor(runtime/v2): untie runtimev2 from the legacy usage of gRPC #22043

Merged
merged 8 commits into from
Oct 3, 2024
Merged

refactor(runtime/v2): untie runtimev2 from the legacy usage of gRPC #22043

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
80657a5
handlers continuation
testinginprod Oct 2, 2024
afc9b06
checkpoint
testinginprod Oct 2, 2024
316578b
checkpoint
testinginprod Oct 2, 2024
2e406ce
lints
testinginprod Oct 2, 2024
4f082de
fix grpc query logic
testinginprod Oct 2, 2024
39633c9
tidy
testinginprod Oct 2, 2024
cc9aab4
Merge branch 'main' into tip/handlers_continuation
testinginprod Oct 2, 2024
a79efec
Apply suggestions from code review
testinginprod Oct 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions runtime/v2/app.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,8 @@ import (
"errors"
"slices"

gogoproto "github.com/cosmos/gogoproto/proto"

runtimev2 "cosmossdk.io/api/cosmos/app/runtime/v2"
appmodulev2 "cosmossdk.io/core/appmodule/v2"
"cosmossdk.io/core/registry"
"cosmossdk.io/core/transaction"
"cosmossdk.io/log"
Expand Down Expand Up @@ -43,9 +42,8 @@ type App[T transaction.Tx] struct {
amino registry.AminoRegistrar
moduleManager *MM[T]

// GRPCMethodsToMessageMap maps gRPC method name to a function that decodes the request
// bytes into a gogoproto.Message, which then can be passed to appmanager.
GRPCMethodsToMessageMap map[string]func() gogoproto.Message
// QueryHandlers defines the query handlers
QueryHandlers map[string]appmodulev2.Handler

storeLoader StoreLoader
}
Expand Down Expand Up @@ -120,6 +118,6 @@ func (a *App[T]) GetAppManager() *appmanager.AppManager[T] {
return a.AppManager
}

func (a *App[T]) GetGPRCMethodsToMessageMap() map[string]func() gogoproto.Message {
return a.GRPCMethodsToMessageMap
func (a *App[T]) GetQueryHandlers() map[string]appmodulev2.Handler {
return a.QueryHandlers
julienrbrt marked this conversation as resolved.
Show resolved Hide resolved
}
156 changes: 88 additions & 68 deletions runtime/v2/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,46 +615,46 @@
}

func registerServices[T transaction.Tx](s appmodulev2.AppModule, app *App[T], registry *protoregistry.Files) error {
c := &configurator{
grpcQueryDecoders: map[string]func() gogoproto.Message{},
stfQueryRouter: app.queryRouterBuilder,
stfMsgRouter: app.msgRouterBuilder,
registry: registry,
err: nil,
}

// case module with services
if services, ok := s.(hasServicesV1); ok {
c := &configurator{
queryHandlers: map[string]appmodulev2.Handler{},
stfQueryRouter: app.queryRouterBuilder,
stfMsgRouter: app.msgRouterBuilder,
registry: registry,
err: nil,
}
Comment on lines +620 to +626
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Simplify struct initialization and avoid field alignment

Per the Uber Go style guide, aligning struct fields using extra spaces is discouraged. Additionally, explicitly initializing the err field to nil is unnecessary since the zero value of an error is nil.

Apply this diff to simplify the struct initialization:

 c := &configurator{
-    queryHandlers:  map[string]appmodulev2.Handler{},
-    stfQueryRouter: app.queryRouterBuilder,
-    stfMsgRouter:   app.msgRouterBuilder,
-    registry:       registry,
-    err:            nil,
+    queryHandlers: map[string]appmodulev2.Handler{},
+    stfQueryRouter: app.queryRouterBuilder,
+    stfMsgRouter: app.msgRouterBuilder,
+    registry: registry,
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
c := &configurator{
queryHandlers: map[string]appmodulev2.Handler{},
stfQueryRouter: app.queryRouterBuilder,
stfMsgRouter: app.msgRouterBuilder,
registry: registry,
err: nil,
}
c := &configurator{
queryHandlers: map[string]appmodulev2.Handler{},
stfQueryRouter: app.queryRouterBuilder,
stfMsgRouter: app.msgRouterBuilder,
registry: registry,
}

if err := services.RegisterServices(c); err != nil {
return fmt.Errorf("unable to register services: %w", err)
}
} else {
// If module not implement RegisterServices, register msg & query handler.
if module, ok := s.(appmodulev2.HasMsgHandlers); ok {
wrapper := stfRouterWrapper{stfRouter: app.msgRouterBuilder}
module.RegisterMsgHandlers(&wrapper)
if wrapper.error != nil {
return fmt.Errorf("unable to register handlers: %w", wrapper.error)
}
}

if module, ok := s.(appmodulev2.HasQueryHandlers); ok {
wrapper := stfRouterWrapper{stfRouter: app.msgRouterBuilder}
module.RegisterQueryHandlers(&wrapper)

for path, decoder := range wrapper.decoders {
app.GRPCMethodsToMessageMap[path] = decoder
}
if c.err != nil {
app.logger.Warn("error registering services", "error", c.err)
}

// merge maps
for path, decoder := range c.queryHandlers {
app.QueryHandlers[path] = decoder
}
Comment on lines +636 to +638

Check warning

Code scanning / CodeQL

Iteration over map Warning

Iteration over map may be a possible source of non-determinism
}

if c.err != nil {
app.logger.Warn("error registering services", "error", c.err)
// if module implements register msg handlers
if module, ok := s.(appmodulev2.HasMsgHandlers); ok {
wrapper := stfRouterWrapper{stfRouter: app.msgRouterBuilder}
module.RegisterMsgHandlers(&wrapper)
if wrapper.error != nil {
return fmt.Errorf("unable to register handlers: %w", wrapper.error)
}
}

// merge maps
for path, decoder := range c.grpcQueryDecoders {
app.GRPCMethodsToMessageMap[path] = decoder
// if module implements register query handlers
if module, ok := s.(appmodulev2.HasQueryHandlers); ok {
wrapper := stfRouterWrapper{stfRouter: app.queryRouterBuilder}
module.RegisterQueryHandlers(&wrapper)

for path, handler := range wrapper.handlers {
app.QueryHandlers[path] = handler
}
Comment on lines +655 to +657

Check warning

Code scanning / CodeQL

Iteration over map Warning

Iteration over map may be a possible source of non-determinism
}

return nil
Expand All @@ -663,9 +663,7 @@
var _ grpc.ServiceRegistrar = (*configurator)(nil)

type configurator struct {
// grpcQueryDecoders is required because module expose queries through gRPC
// this provides a way to route to modules using gRPC.
grpcQueryDecoders map[string]func() gogoproto.Message
queryHandlers map[string]appmodulev2.Handler

stfQueryRouter *stf.MsgRouterBuilder
stfMsgRouter *stf.MsgRouterBuilder
Expand Down Expand Up @@ -697,61 +695,59 @@
func (c *configurator) registerQueryHandlers(sd *grpc.ServiceDesc, ss interface{}) error {
for _, md := range sd.Methods {
// TODO(tip): what if a query is not deterministic?
requestFullName, err := registerMethod(c.stfQueryRouter, sd, md, ss)

handler, err := grpcHandlerToAppModuleHandler(sd, md, ss)
if err != nil {
return fmt.Errorf("unable to register query handler %s.%s: %w", sd.ServiceName, md.MethodName, err)
return fmt.Errorf("unable to make a appmodulev2.HandlerFunc from gRPC handler (%s, %s): %w", sd.ServiceName, md.MethodName, err)
}

// register gRPC query method.
typ := gogoproto.MessageType(requestFullName)
if typ == nil {
return fmt.Errorf("unable to find message in gogotype registry: %w", err)
}
decoderFunc := func() gogoproto.Message {
return reflect.New(typ.Elem()).Interface().(gogoproto.Message)
// register to stf query router.
err = c.stfQueryRouter.RegisterHandler(gogoproto.MessageName(handler.MakeMsg()), handler.Func)
if err != nil {
return fmt.Errorf("unable to register handler to stf router (%s, %s): %w", sd.ServiceName, md.MethodName, err)
}
methodName := fmt.Sprintf("/%s/%s", sd.ServiceName, md.MethodName)
c.grpcQueryDecoders[methodName] = decoderFunc

// register query handler using the same mapping used in stf
c.queryHandlers[gogoproto.MessageName(handler.MakeMsg())] = handler
}
return nil
}

func (c *configurator) registerMsgHandlers(sd *grpc.ServiceDesc, ss interface{}) error {
for _, md := range sd.Methods {
_, err := registerMethod(c.stfMsgRouter, sd, md, ss)
handler, err := grpcHandlerToAppModuleHandler(sd, md, ss)
if err != nil {
return err
}
err = c.stfMsgRouter.RegisterHandler(gogoproto.MessageName(handler.MakeMsg()), handler.Func)
if err != nil {
return fmt.Errorf("unable to register msg handler %s.%s: %w", sd.ServiceName, md.MethodName, err)
}
}
return nil
}

// requestFullNameFromMethodDesc returns the fully-qualified name of the request message of the provided service's method.
func requestFullNameFromMethodDesc(sd *grpc.ServiceDesc, method grpc.MethodDesc) (protoreflect.FullName, error) {
methodFullName := protoreflect.FullName(fmt.Sprintf("%s.%s", sd.ServiceName, method.MethodName))
desc, err := gogoproto.HybridResolver.FindDescriptorByName(methodFullName)
if err != nil {
return "", fmt.Errorf("cannot find method descriptor %s", methodFullName)
}
methodDesc, ok := desc.(protoreflect.MethodDescriptor)
if !ok {
return "", fmt.Errorf("invalid method descriptor %s", methodFullName)
}
return methodDesc.Input().FullName(), nil
}

func registerMethod(
stfRouter *stf.MsgRouterBuilder,
// grpcHandlerToAppModuleHandler converts a gRPC handler into an appmodulev2.HandlerFunc.
func grpcHandlerToAppModuleHandler(
sd *grpc.ServiceDesc,
md grpc.MethodDesc,
ss interface{},
) (string, error) {
requestName, err := requestFullNameFromMethodDesc(sd, md)
) (appmodulev2.Handler, error) {
requestName, responseName, err := requestFullNameFromMethodDesc(sd, md)
if err != nil {
return "", err
return appmodulev2.Handler{}, err
}

return string(requestName), stfRouter.RegisterHandler(string(requestName), func(
requestTyp := gogoproto.MessageType(string(requestName))
if requestTyp == nil {
return appmodulev2.Handler{}, fmt.Errorf("no proto message found for %s", requestName)
}
responseTyp := gogoproto.MessageType(string(responseName))
if responseTyp == nil {
return appmodulev2.Handler{}, fmt.Errorf("no proto message found for %s", responseName)
}

handlerFunc := func(
ctx context.Context,
msg transaction.Msg,
) (resp transaction.Msg, err error) {
Expand All @@ -760,7 +756,17 @@
return nil, err
}
return res.(transaction.Msg), nil
})
}

return appmodulev2.Handler{
Func: handlerFunc,
MakeMsg: func() transaction.Msg {
return reflect.New(requestTyp.Elem()).Interface().(transaction.Msg)
},
MakeMsgResp: func() transaction.Msg {
return reflect.New(responseTyp.Elem()).Interface().(transaction.Msg)
},
Comment on lines +763 to +768
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Add checks for type assertions in MakeMsg and MakeMsgResp

The type assertions in MakeMsg and MakeMsgResp may cause panics if the created instances do not implement transaction.Msg. Consider adding checks to handle potential type assertion failures.

Apply this diff to safely handle the type assertions:

 MakeMsg: func() transaction.Msg {
-	return reflect.New(requestTyp.Elem()).Interface().(transaction.Msg)
+	msg, ok := reflect.New(requestTyp.Elem()).Interface().(transaction.Msg)
+	if !ok {
+		return nil
+	}
+	return msg
},
MakeMsgResp: func() transaction.Msg {
-	return reflect.New(responseTyp.Elem()).Interface().(transaction.Msg)
+	msgResp, ok := reflect.New(responseTyp.Elem()).Interface().(transaction.Msg)
+	if !ok {
+		return nil
+	}
+	return msgResp
},
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
MakeMsg: func() transaction.Msg {
return reflect.New(requestTyp.Elem()).Interface().(transaction.Msg)
},
MakeMsgResp: func() transaction.Msg {
return reflect.New(responseTyp.Elem()).Interface().(transaction.Msg)
},
MakeMsg: func() transaction.Msg {
msg, ok := reflect.New(requestTyp.Elem()).Interface().(transaction.Msg)
if !ok {
return nil
}
return msg
},
MakeMsgResp: func() transaction.Msg {
msgResp, ok := reflect.New(responseTyp.Elem()).Interface().(transaction.Msg)
if !ok {
return nil
}
return msgResp
},

}, nil
}

func noopDecoder(_ interface{}) error { return nil }
Expand All @@ -776,6 +782,20 @@
}
}

// requestFullNameFromMethodDesc returns the fully-qualified name of the request message and response of the provided service's method.
func requestFullNameFromMethodDesc(sd *grpc.ServiceDesc, method grpc.MethodDesc) (protoreflect.FullName, protoreflect.FullName, error) {
methodFullName := protoreflect.FullName(fmt.Sprintf("%s.%s", sd.ServiceName, method.MethodName))
desc, err := gogoproto.HybridResolver.FindDescriptorByName(methodFullName)
if err != nil {
return "", "", fmt.Errorf("cannot find method descriptor %s", methodFullName)
}
methodDesc, ok := desc.(protoreflect.MethodDescriptor)
if !ok {
return "", "", fmt.Errorf("invalid method descriptor %s", methodFullName)
}
return methodDesc.Input().FullName(), methodDesc.Output().FullName(), nil
}

// defaultMigrationsOrder returns a default migrations order: ascending alphabetical by module name,
// except x/auth which will run last, see:
// https://github.com/cosmos/cosmos-sdk/issues/10591
Expand Down Expand Up @@ -815,7 +835,7 @@

error error

decoders map[string]func() gogoproto.Message
handlers map[string]appmodulev2.Handler
}

func (s *stfRouterWrapper) RegisterHandler(handler appmodulev2.Handler) {
Expand All @@ -831,7 +851,7 @@

// also make the decoder
if s.error == nil {
s.decoders = map[string]func() gogoproto.Message{}
s.handlers = map[string]appmodulev2.Handler{}
}
s.decoders[requestName] = handler.MakeMsg
s.handlers[requestName] = handler
Comment on lines +854 to +856
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Avoid re-initializing the handlers map to prevent overwriting

The handlers map is being re-initialized inside the RegisterHandler method whenever s.error == nil, which can cause previously registered handlers to be lost. Initialize the handlers map only once, preferably when the stfRouterWrapper is created, or check if it is nil before initializing.

Apply this diff to fix the issue:

 if s.error == nil {
-	s.handlers = map[string]appmodulev2.Handler{}
+	if s.handlers == nil {
+		s.handlers = map[string]appmodulev2.Handler{}
+	}
 }
 s.handlers[requestName] = handler
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
s.handlers = map[string]appmodulev2.Handler{}
}
s.decoders[requestName] = handler.MakeMsg
s.handlers[requestName] = handler
if s.error == nil {
if s.handlers == nil {
s.handlers = map[string]appmodulev2.Handler{}
}
}
s.handlers[requestName] = handler

}
14 changes: 7 additions & 7 deletions runtime/v2/module.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,13 +127,13 @@ func ProvideAppBuilder[T transaction.Tx](

msgRouterBuilder := stf.NewMsgRouterBuilder()
app := &App[T]{
storeKeys: nil,
interfaceRegistrar: interfaceRegistrar,
amino: amino,
msgRouterBuilder: msgRouterBuilder,
queryRouterBuilder: stf.NewMsgRouterBuilder(), // TODO dedicated query router
GRPCMethodsToMessageMap: map[string]func() proto.Message{},
storeLoader: DefaultStoreLoader,
storeKeys: nil,
interfaceRegistrar: interfaceRegistrar,
amino: amino,
msgRouterBuilder: msgRouterBuilder,
queryRouterBuilder: stf.NewMsgRouterBuilder(), // TODO dedicated query router
QueryHandlers: map[string]appmodulev2.Handler{},
storeLoader: DefaultStoreLoader,
}
appBuilder := &AppBuilder[T]{app: app}

Expand Down
35 changes: 29 additions & 6 deletions server/v2/api/grpc/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,18 @@ import (
"net"
"slices"
"strconv"
"strings"
"sync"

"github.com/cosmos/gogoproto/proto"
gogoproto "github.com/cosmos/gogoproto/proto"
"github.com/spf13/pflag"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/reflect/protoreflect"

appmodulev2 "cosmossdk.io/core/appmodule/v2"
"cosmossdk.io/core/transaction"
"cosmossdk.io/log"
serverv2 "cosmossdk.io/server/v2"
Expand Down Expand Up @@ -53,7 +57,7 @@ func (s *Server[T]) Init(appI serverv2.AppI[T], cfg map[string]any, logger log.L
return fmt.Errorf("failed to unmarshal config: %w", err)
}
}
methodsMap := appI.GetGPRCMethodsToMessageMap()
methodsMap := appI.GetQueryHandlers()

grpcSrv := grpc.NewServer(
grpc.ForceServerCodec(newProtoCodec(appI.InterfaceRegistry()).GRPCCodec()),
Expand All @@ -80,21 +84,40 @@ func (s *Server[T]) StartCmdFlags() *pflag.FlagSet {
return flags
}

func makeUnknownServiceHandler(messageMap map[string]func() proto.Message, querier interface {
Query(ctx context.Context, version uint64, msg proto.Message) (proto.Message, error)
func makeUnknownServiceHandler(handlers map[string]appmodulev2.Handler, querier interface {
Query(ctx context.Context, version uint64, msg gogoproto.Message) (gogoproto.Message, error)
},
) grpc.StreamHandler {
getRegistry := sync.OnceValues(gogoproto.MergedRegistry)
testinginprod marked this conversation as resolved.
Show resolved Hide resolved

return func(srv any, stream grpc.ServerStream) error {
method, ok := grpc.MethodFromServerStream(stream)
if !ok {
return status.Error(codes.InvalidArgument, "unable to get method")
}
makeMsg, exists := messageMap[method]
// if this fails we cannot serve queries anymore...
registry, err := getRegistry()
if err != nil {
return fmt.Errorf("failed to get registry: %w", err)
}
fullName := protoreflect.FullName(strings.ReplaceAll(method, "/", "."))
// get descriptor from the invoke method
desc, err := registry.FindDescriptorByName(fullName)
if err != nil {
return fmt.Errorf("failed to find descriptor %s: %w", method, err)
Comment on lines +105 to +107
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Return appropriate gRPC error when descriptor is not found

Instead of returning a generic error, consider returning a gRPC status error to provide clearer feedback to the client when the descriptor is not found.

Apply this diff to modify the error handling:

			if err != nil {
-				return fmt.Errorf("failed to find descriptor %s: %w", method, err)
+				return status.Errorf(codes.Unimplemented, "method %s not found: %v", method, err)
			}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
desc, err := registry.FindDescriptorByName(fullName)
if err != nil {
return fmt.Errorf("failed to find descriptor %s: %w", method, err)
desc, err := registry.FindDescriptorByName(fullName)
if err != nil {
return status.Errorf(codes.Unimplemented, "method %s not found: %v", method, err)

}
md, ok := desc.(protoreflect.MethodDescriptor)
if !ok {
return fmt.Errorf("%s is not a method", method)
}
// find handler
handler, exists := handlers[string(md.Input().FullName())]
if !exists {
return status.Errorf(codes.Unimplemented, "gRPC method %s is not handled", method)
}

for {
req := makeMsg()
req := handler.MakeMsg()
err := stream.RecvMsg(req)
if err != nil {
if errors.Is(err, io.EOF) {
Expand Down
Loading
Loading