bump grpc version (#44)

This merge updates our gRPC dependency.
To achieve this, there were some deprecated portions of code to move away from, and others that had completely broken.

The Data field on StatsHandler structs was deprecated for quite some time, and is now removed.
Additionally, Codec was deprecated, and stopped working, it has been replaced with CodecV2.
The new version of the gRPC generated code now passes StaticMethodCallOption as an option to the clients, so we needed to account for that, as our internal methods did not allow for any call options.
The types passed the generated server methods have changed, so the old mocks didn't work, and we had to mock a different type in the tests.

Co-Authored-By Sam Childs <[email protected]>

.github/workflows/main.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Setup Golang with cache
         uses: magnetikonline/action-golang-cache@v3
         with:
-          go-version: 1.19
+          go-version: 1.22.4
 
       - run: go version
 

.mockery.yaml

@@ -12,7 +12,7 @@ packages:
     config:
       dir: "gen/grpc/mocks"
       outpkg: "mocks"
-      include-regex: ".*Interceptor"
+      include-regex: ".*Interceptor|.*BidiStreamingServer"
   google.golang.org/grpc/stats: 
     config:
       dir: "gen/grpc/stats/mocks"

Taskfile.yml

@@ -3,7 +3,7 @@ method: timestamp
 run: once
 
 vars:
-  MOCKERY: github.com/vektra/mockery/v2@v2.34.2
+  MOCKERY: github.com/vektra/mockery/v2@v2.45.0
   EXHAUSTIVE: github.com/nishanths/exhaustive/cmd/[email protected]
 
 tasks:

client.go

@@ -13,14 +13,15 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/encoding"
 	"google.golang.org/grpc/encoding/proto"
+	"google.golang.org/grpc/mem"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
 )
 
 type ClientConn struct {
 	mp *client.RpcMultiplexer
 
-	codec encoding.Codec
+	codec encoding.CodecV2
 
 	unaryInterceptor  grpc.UnaryClientInterceptor
 	streamInterceptor grpc.StreamClientInterceptor
@@ -35,7 +36,7 @@ var _ grpc.ClientConnInterface = (*ClientConn)(nil)
 func NewClientConn(conn RpcReadWriter, source, dest string, opts ...DialOption) *ClientConn {
 	cc := ClientConn{
 		mp:            client.NewRpcMultiplexer(conn),
-		codec:         encoding.GetCodec(proto.Name),
+		codec:         encoding.GetCodecV2(proto.Name),
 		sourceAddress: source,
 		destAddress:   dest,
 	}
@@ -86,8 +87,14 @@ func (cc *ClientConn) invoke(
 	reply interface{},
 	opts ...grpc.CallOption,
 ) error {
-	if len(opts) > 0 {
-		log.Panic().Msg("Invoke opts unsupported")
+	if len(opts) > 1 {
+		log.Panic().Msg("Invoke: opts unsupported")
+	} else if len(opts) == 1 {
+		switch opts[0].(type) {
+		case grpc.StaticMethodCallOption:
+		default:
+			log.Panic().Msg("Invoke: opts unsupported")
+		}
 	}
 
 	var err error
@@ -114,10 +121,10 @@ func (cc *ClientConn) invoke(
 			FullMethod: method,
 			Header:     mdHeaders,
 		})
+
 		sh.HandleRPC(ctx, &stats.OutPayload{
 			Client:   true,
 			Payload:  args,
-			Data:     body,
 			Length:   len(body),
 			SentTime: time.Now(),
 		})
@@ -133,7 +140,7 @@ func (cc *ClientConn) invoke(
 			Destination: cc.destAddress,
 		},
 		&goatorepo.Body{
-			Data: body,
+			Data: body.Materialize(),
 		},
 		cc.statsHandlers,
 	)
@@ -143,7 +150,10 @@ func (cc *ClientConn) invoke(
 		return err
 	}
 
-	err = cc.codec.Unmarshal(replyBody.GetData(), reply)
+	buf := mem.NewBuffer(&replyBody.Data, nil)
+	bs := mem.BufferSlice{buf}
+
+	err = cc.codec.Unmarshal(bs, reply)
 	if err != nil {
 		log.Error().Err(err).Msg("Invoke Unmarshal")
 	}
@@ -152,7 +162,6 @@ func (cc *ClientConn) invoke(
 		sh.HandleRPC(ctx, &stats.InPayload{
 			Client:   true,
 			Payload:  reply,
-			Data:     replyBody.GetData(),
 			Length:   len(replyBody.GetData()),
 			RecvTime: time.Now(),
 		})
@@ -192,8 +201,14 @@ func (cc *ClientConn) newStream(
 	method string,
 	opts ...grpc.CallOption,
 ) (grpc.ClientStream, error) {
-	if len(opts) > 0 {
+	if len(opts) > 1 {
 		log.Panic().Msg("NewStream: opts unsupported")
+	} else if len(opts) == 1 {
+		switch opts[0].(type) {
+		case grpc.StaticMethodCallOption:
+		default:
+			log.Panic().Msg("NewStream: opts unsupported")
+		}
 	}
 
 	id, rw, teardown, err := cc.mp.NewStreamReadWriter(ctx)