From 983c78d4333f2c94b2bf86b37c6abd9ac0700d48 Mon Sep 17 00:00:00 2001 From: Vaughan Andrews Date: Fri, 4 Sep 2026 09:28:32 -0700 Subject: [PATCH] [translation]: Add a library for substituting one gRPC method for another Some upstreams do not serve a method under the name a caller knows it by. The proxy can forward a method or refuse it, but it has no way to answer one call with a different one, which is what an upstream that exposes the same capability under another service requires. Add internal/translation. A Translation names the inbound method, the method that stands in for it, and the conversions between their message types; a Registry indexes them by inbound method; and a unary client interceptor converts the request, sends it under the substituted method, and folds the reply back into the message the caller is waiting on. Nothing else on the connection needs to know a substitution happened. Adapt builds a Translation from two typed conversions, so a mapping is written against concrete message types and never asserts on proto.Message itself. The response conversion is given the original request as well as the upstream reply, because a request field the upstream has no equivalent for can only be honoured on the way back. WithHeader stamps metadata the substituted API requires, set rather than appended, and only on a call that was actually substituted. Via sends the substituted call over a different connection, for an upstream method the original connection's service does not serve at all. It leaves the interceptor chain at that point, which is why callers install this innermost: everything above then sees the method and message types the caller asked for. No translations are registered yet, so nothing changes. --- internal/cloud/translation/doc.go | 23 ++ internal/cloud/translation/interceptor.go | 122 +++++++ .../cloud/translation/interceptor_test.go | 306 ++++++++++++++++++ internal/cloud/translation/translation.go | 213 ++++++++++++ .../cloud/translation/translation_test.go | 113 +++++++ 5 files changed, 777 insertions(+) create mode 100644 internal/cloud/translation/doc.go create mode 100644 internal/cloud/translation/interceptor.go create mode 100644 internal/cloud/translation/interceptor_test.go create mode 100644 internal/cloud/translation/translation.go create mode 100644 internal/cloud/translation/translation_test.go diff --git a/internal/cloud/translation/doc.go b/internal/cloud/translation/doc.go new file mode 100644 index 0000000..5bde9d1 --- /dev/null +++ b/internal/cloud/translation/doc.go @@ -0,0 +1,23 @@ +// Package translation rewrites one gRPC method call into another on the hop to +// the upstream. +// +// A [Translation] pairs an inbound method with the upstream method that stands +// in for it, plus the conversions between their request and response types. A +// [Registry] holds the set of them, and [DialOptions] installs them on a +// connection: a call whose method is registered is converted, sent under the +// upstream method, and converted back before the caller sees a reply. +// Everything else is passed straight through. +// +// It lives under internal/cloud because Temporal Cloud is what needs it - Cloud +// serves some methods only from its control plane, under another service - and +// because "translation" unqualified already means namespace translation +// elsewhere in the proxy. That one rewrites names inside a message and leaves +// the method alone; this one replaces the call. The two compose: install this +// innermost, so namespace translation, payload codecs, and the reflective +// forwarder all keep seeing the method and message types the caller asked for. +// +// The mechanism itself knows nothing about Cloud, and is kept separate from the +// parent package so that using [cloud.IsEndpoint] or [cloud.ValidateNamespace] +// does not pull gRPC and protobuf into a caller that only wanted to check a +// name. +package translation diff --git a/internal/cloud/translation/interceptor.go b/internal/cloud/translation/interceptor.go new file mode 100644 index 0000000..4f85eda --- /dev/null +++ b/internal/cloud/translation/interceptor.go @@ -0,0 +1,122 @@ +package translation + +import ( + "context" + + "google.golang.org/grpc" + "google.golang.org/protobuf/proto" + + "github.com/temporalio/temporal-proxy/internal/rpc" +) + +// Option configures the interceptor [DialOptions] installs. +type Option func(*options) + +type options struct { + via grpc.ClientConnInterface +} + +// Via sends a translated call over cc instead of continuing down the chain to +// the connection the interceptor is installed on. +// +// The upstream method belongs to a different service, which the connection that +// received the call does not serve: the caller asked a Temporal Service for +// ListNamespaces, and only Temporal Cloud's control plane can answer it. Where +// that service lives is as fixed as the conversions themselves, so the +// translation carries the connection rather than the request being routed to it, +// and a request reaching any upstream is answered the same way. +func Via(cc grpc.ClientConnInterface) Option { + return func(o *options) { o.via = cc } +} + +// DialOptions returns the dial options that install method translation on an +// outbound connection. Callers fold them into the dial options for the upstream +// connection, last, so translation is the innermost interceptor: every other +// interceptor on the chain then sees the method and message types the caller +// asked for rather than the substitute sent upstream. +// +// The result is a slice though it holds a single option today. A [Translation] +// substitutes a unary method, so a unary interceptor is all there is to install; +// translating a streaming method would add a stream interceptor beside it, the +// way the namespace and Cloud-namespace helpers in internal/proxy already pair +// the two. Keeping the slice means that arrives without changing this signature +// or the call sites, which already spread the result. +func DialOptions(r *Registry, opts ...Option) []grpc.DialOption { + return []grpc.DialOption{grpc.WithChainUnaryInterceptor(unaryClientInterceptor(r, opts...))} +} + +// unaryClientInterceptor returns a unary client interceptor that replaces a call +// to a method r translates with a call to the method it translates onto: the +// request is converted, invoked under the upstream method, and the upstream's +// reply is folded into the reply the caller allocated. A method r does not +// translate is invoked unchanged, as is a call whose request or reply is not a +// proto message. Any headers the translation declares are stamped on the +// substituted call only. +// +// An upstream error is returned as it arrived, so the caller sees the upstream's +// status rather than a translated one. A conversion that fails becomes Internal +// via [rpc.StatusError]: the mapping is compiled in, so a failure there is a +// proxy bug rather than anything the caller did. +func unaryClientInterceptor(r *Registry, opts ...Option) grpc.UnaryClientInterceptor { + o := &options{} + for _, opt := range opts { + opt(o) + } + + return func( + ctx context.Context, + method string, + req, reply any, + cc *grpc.ClientConn, + invoker grpc.UnaryInvoker, + callOpts ...grpc.CallOption, + ) error { + t, ok := r.Lookup(method) + if !ok { + return invoker(ctx, method, req, reply, cc, callOpts...) + } + + in, inOK := req.(proto.Message) + out, outOK := reply.(proto.Message) + if !inOK || !outOK { + // Nothing to convert against. The forwarder types every message from the + // proto registry, so this only happens on a hand-rolled call; forwarding + // it unchanged is closer to right than failing it. + return invoker(ctx, method, req, reply, cc, callOpts...) + } + + upReq, err := t.request(in) + if err != nil { + return rpc.StatusError("translation: adapting the request failed", err) + } + + upReply := t.reply() + if err := o.invoke(t.stamp(ctx), t, upReq, upReply, cc, invoker, callOpts...); err != nil { + return err + } + + if err := t.response(in, upReply, out); err != nil { + return rpc.StatusError("translation: adapting the reply failed", err) + } + + return nil + } +} + +// invoke sends the substituted call: over the connection Via named when there is +// one, and otherwise down the chain to the connection the interceptor was +// installed on. +func (o *options) invoke( + ctx context.Context, + t *Translation, + req, reply any, + cc *grpc.ClientConn, + invoker grpc.UnaryInvoker, + callOpts ...grpc.CallOption, +) error { + if o.via != nil { + return o.via.Invoke(ctx, t.to, req, reply, callOpts...) + } + + return invoker(ctx, t.to, req, reply, cc, callOpts...) +} diff --git a/internal/cloud/translation/interceptor_test.go b/internal/cloud/translation/interceptor_test.go new file mode 100644 index 0000000..c861e52 --- /dev/null +++ b/internal/cloud/translation/interceptor_test.go @@ -0,0 +1,306 @@ +package translation_test + +import ( + "context" + "errors" + "net" + "sync" + "testing" + + "github.com/stretchr/testify/require" + workflowservice "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/cloud/translation" +) + +// systemInfoService is a fake upstream serving the method the test translation +// substitutes onto. It records what it was called with, so a test can tell a +// substituted call from one that was forwarded unchanged. +type systemInfoService struct { + workflowservice.UnimplementedWorkflowServiceServer + + lis net.Listener + version string + err error + + // mu guards the recorded state, which the serving goroutine writes while the + // test reads. + mu sync.Mutex + md metadata.MD + called bool +} + +func TestDialOptionsSubstitutesTheUpstreamCall(t *testing.T) { + t.Parallel() + + upstream := newSystemInfoService(t) + upstream.version = "1.2.3" + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + // The reply is built from both halves: "payments" came from the caller's + // request and "1.2.3" from the substituted call's reply, so one assertion + // covers the substitution and the conversion in both directions. + reply := &workflowservice.DescribeNamespaceResponse{} + err := cc.Invoke(t.Context(), fromMethod, &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply) + require.NoError(t, err) + + require.Equal(t, "payments@1.2.3", reply.GetNamespaceInfo().GetName()) + require.True(t, upstream.wasCalled(), "the substituted method must have been invoked") +} + +func TestDialOptionsPassesUntranslatedMethodsThrough(t *testing.T) { + t.Parallel() + + // A method the registry does not translate reaches the upstream under its own + // name, which the fake does not implement. + const untranslated = "/temporal.api.workflowservice.v1.WorkflowService/ListNamespaces" + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + err := cc.Invoke( + t.Context(), untranslated, + &workflowservice.ListNamespacesRequest{}, &workflowservice.ListNamespacesResponse{}, + ) + require.Equal(t, codes.Unimplemented, status.Code(err)) + require.False(t, upstream.wasCalled(), "and must not be substituted onto one that is implemented") +} + +func TestDialOptionsReturnsTheUpstreamError(t *testing.T) { + t.Parallel() + + upstream := newSystemInfoService(t) + upstream.err = status.Error(codes.PermissionDenied, "no") + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + err := cc.Invoke( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, + ) + require.Equal(t, codes.PermissionDenied, status.Code(err), "the caller sees the upstream's own status") + require.Equal(t, "no", status.Convert(err).Message()) +} + +func TestDialOptionsReportsConversionFailureAsInternal(t *testing.T) { + t.Parallel() + + boom := errors.New("boom") + r, err := translation.NewRegistry(translation.Adapt( + fromMethod, toMethod, + func(*workflowservice.DescribeNamespaceRequest) (*workflowservice.GetSystemInfoRequest, error) { + return nil, boom + }, + okResponse, + )) + require.NoError(t, err) + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(r)...) + + err = cc.Invoke( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, + ) + require.Equal(t, codes.Internal, status.Code(err), "a compiled-in mapping that fails is a proxy bug") + require.ErrorContains(t, err, "boom") + require.False(t, upstream.wasCalled(), "and the upstream is never reached") +} + +func TestDialOptionsReportsAMismatchedReplyAsInternal(t *testing.T) { + t.Parallel() + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + // A reply of the wrong type for the registered translation: the request + // converts, the substituted call succeeds, and folding the reply back fails. + err := cc.Invoke( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.ListNamespacesResponse{}, + ) + require.Equal(t, codes.Internal, status.Code(err)) + require.ErrorContains(t, err, "wanted a temporal.api.workflowservice.v1.DescribeNamespaceResponse reply") +} + +func TestDialOptionsReportsAMismatchedRequestAsInternal(t *testing.T) { + t.Parallel() + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + // A proto request, but not the one this translation converts. The registry + // matched on the method, so the mismatch can only be caught by the conversion. + err := cc.Invoke( + t.Context(), fromMethod, + &workflowservice.ListNamespacesRequest{}, &workflowservice.DescribeNamespaceResponse{}, + ) + require.Equal(t, codes.Internal, status.Code(err)) + require.ErrorContains(t, err, "wanted a temporal.api.workflowservice.v1.DescribeNamespaceRequest request") + require.False(t, upstream.wasCalled(), "and the upstream is never reached") +} + +func TestDialOptionsForwardsNonProtoCallsUnchanged(t *testing.T) { + t.Parallel() + + // Nothing the proxy does reaches here - the reflective forwarder types every + // message from the proto registry, and a generated client only ever passes + // proto messages - so this drives the interceptor directly with values no + // codec can marshal. What matters is that the call is left alone rather than + // substituted: it fails as an unmarshalable call to the method the caller + // named, not as a translated one. + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(testRegistry(t))...) + + req, reply := 42, 0 + err := cc.Invoke(t.Context(), fromMethod, req, &reply) + // gRPC's own codec rejects it, which is what "left alone" looks like from + // here: had the call been translated, it would have failed in a conversion + // with a "wanted a ... request" message instead. + require.ErrorContains(t, err, "error while marshaling") + require.False(t, upstream.wasCalled(), "and nothing was substituted onto the upstream method") +} + +func TestDialOptionsStampsHeadersOnTheSubstitutedCallOnly(t *testing.T) { + t.Parallel() + + const header = "x-api-version" + + r, err := translation.NewRegistry( + translation.Adapt(fromMethod, toMethod, okRequest, okResponse).WithHeader(header, "pinned"), + ) + require.NoError(t, err) + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(r)...) + + // The caller sent its own value, which the substituted API cannot be assumed + // to understand; the translation pins what its conversions were written + // against, and must leave exactly one value rather than appending a second. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(header, "caller")) + err = cc.Invoke(ctx, fromMethod, &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}) + require.NoError(t, err) + require.Equal(t, []string{"pinned"}, upstream.metadata().Get(header)) +} + +func TestDialOptionsLeavesAnUntranslatedCallsHeadersAlone(t *testing.T) { + t.Parallel() + + const header = "x-api-version" + + // The same translation, but invoked as the substituted method directly - a + // caller speaking that API already, whose own version must travel. + r, err := translation.NewRegistry( + translation.Adapt(fromMethod, toMethod, okRequest, okResponse).WithHeader(header, "pinned"), + ) + require.NoError(t, err) + + upstream := newSystemInfoService(t) + cc := dial(t, upstream, translation.DialOptions(r)...) + + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(header, "caller")) + err = cc.Invoke(ctx, toMethod, &workflowservice.GetSystemInfoRequest{}, &workflowservice.GetSystemInfoResponse{}) + require.NoError(t, err) + require.Equal(t, []string{"caller"}, upstream.metadata().Get(header)) +} + +func TestViaSendsTheSubstitutedCallElsewhere(t *testing.T) { + t.Parallel() + + // Via is what lets a translation answer over a connection other than the one + // it is installed on, for an upstream method the original connection's service + // does not serve at all. + elsewhere := newSystemInfoService(t) + elsewhere.version = "9.9.9" + elsewhereConn := dial(t, elsewhere) + + installed := newSystemInfoService(t) + cc := dial(t, installed, translation.DialOptions(testRegistry(t), translation.Via(elsewhereConn))...) + + reply := &workflowservice.DescribeNamespaceResponse{} + err := cc.Invoke(t.Context(), fromMethod, &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply) + require.NoError(t, err) + + require.Equal(t, "payments@9.9.9", reply.GetNamespaceInfo().GetName()) + require.True(t, elsewhere.wasCalled(), "the substituted call goes to the Via connection") + require.False(t, installed.wasCalled(), "and leaves the chain rather than continuing down it") +} + +// testRegistry holds the one test translation. +func testRegistry(t *testing.T) *translation.Registry { + t.Helper() + + r, err := translation.NewRegistry(translation.Adapt(fromMethod, toMethod, okRequest, okResponse)) + require.NoError(t, err) + + return r +} + +// dial returns a client connection to up carrying opts, which are the dial +// options under test. Passing none gives a plain connection, for use as a Via +// target. +func dial(t *testing.T, up *systemInfoService, opts ...grpc.DialOption) *grpc.ClientConn { + t.Helper() + + cc, err := grpc.NewClient( + up.lis.Addr().String(), + append([]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, opts...)..., + ) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + return cc +} + +// newSystemInfoService starts the fake on a loopback port and stops it when the +// test ends. +func newSystemInfoService(t *testing.T) *systemInfoService { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + svc := &systemInfoService{lis: lis} + + svr := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(svr, svc) + go func() { _ = svr.Serve(lis) }() + t.Cleanup(svr.Stop) + + return svc +} + +func (s *systemInfoService) GetSystemInfo( + ctx context.Context, _ *workflowservice.GetSystemInfoRequest, +) (*workflowservice.GetSystemInfoResponse, error) { + md, _ := metadata.FromIncomingContext(ctx) + + s.mu.Lock() + s.md = md + s.called = true + s.mu.Unlock() + + if s.err != nil { + return nil, s.err + } + + return &workflowservice.GetSystemInfoResponse{ServerVersion: s.version}, nil +} + +func (s *systemInfoService) wasCalled() bool { + s.mu.Lock() + defer s.mu.Unlock() + + return s.called +} + +func (s *systemInfoService) metadata() metadata.MD { + s.mu.Lock() + defer s.mu.Unlock() + + return s.md.Copy() +} diff --git a/internal/cloud/translation/translation.go b/internal/cloud/translation/translation.go new file mode 100644 index 0000000..f51fb1a --- /dev/null +++ b/internal/cloud/translation/translation.go @@ -0,0 +1,213 @@ +package translation + +import ( + "context" + "fmt" + + "google.golang.org/grpc/metadata" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + + "github.com/temporalio/temporal-proxy/internal/rpc" +) + +type ( + // Translation stands one unary method in for another: it converts the + // caller's request into the upstream's request type, allocates the reply the + // upstream will fill, and folds that reply back into the message type the + // caller is waiting on. Build one with [Adapt] rather than by hand, so the + // conversions are written against concrete message types. A Translation holds + // no per-call state and is safe for concurrent use. + Translation struct { + from, to string + request func(req proto.Message) (proto.Message, error) + response func(req, upstream, reply proto.Message) error + reply func() proto.Message + headers map[string]string + } + + // Registry is the set of translations an interceptor consults, keyed by the + // inbound full method. It is fixed once built and safe for concurrent use; a + // nil Registry translates nothing, so a caller with none to install can pass + // one straight through. + Registry struct { + byMethod map[string]*Translation + } +) + +// Adapt builds a [Translation] from from onto to out of two typed conversions. +// request converts the caller's request into the upstream's; response folds the +// upstream's reply into the caller's, and is given the original request too, +// since a field the upstream has no equivalent for (a filter, say) can only be +// honoured on the way back. Both message types are inferred from the +// conversions, so a mapping never asserts on proto.Message itself. +func Adapt[Req, UpReq, UpResp, Resp proto.Message]( + from, to string, + request func(Req) (UpReq, error), + response func(Req, UpResp, Resp) error, +) *Translation { + return &Translation{ + from: from, + to: to, + reply: func() proto.Message { + return newMessage[UpResp]() + }, + request: func(m proto.Message) (proto.Message, error) { + req, ok := m.(Req) + if !ok { + return nil, fmt.Errorf("translation: %s wanted a %s request, got %T", from, nameOf[Req](), m) + } + + return request(req) + }, + response: func(m, up, out proto.Message) error { + req, ok := m.(Req) + if !ok { + return fmt.Errorf("translation: %s wanted a %s request, got %T", from, nameOf[Req](), m) + } + + upstream, ok := up.(UpResp) + if !ok { + return fmt.Errorf("translation: %s wanted a %s reply, got %T", to, nameOf[UpResp](), up) + } + + reply, ok := out.(Resp) + if !ok { + return fmt.Errorf("translation: %s wanted a %s reply, got %T", from, nameOf[Resp](), out) + } + + return response(req, upstream, reply) + }, + } +} + +// WithHeader stamps key: value on the substituted call and returns t, so a +// mapping can declare the dialect the upstream method needs alongside the +// conversions themselves. It replaces any value the caller sent rather than +// adding to it: the caller did not ask for this upstream method and cannot know +// what its API expects, so its own header is not intent worth preserving. A +// caller invoking that API directly is forwarded untranslated and keeps its +// header. +// +// Headers travel only on a call this translation substituted; a method the +// registry does not translate is untouched. +func (t *Translation) WithHeader(key, value string) *Translation { + if t.headers == nil { + t.headers = make(map[string]string, 1) + } + + t.headers[key] = value + + return t +} + +// NewRegistry indexes ts by the method each translates from. It rejects a nil +// entry, a method name that is not a gRPC full method, a translation onto +// itself, and two translations of the same inbound method, so a mapping mistake +// surfaces at construction rather than on the first request that hits it. +func NewRegistry(ts ...*Translation) (*Registry, error) { + byMethod := make(map[string]*Translation, len(ts)) + for i, t := range ts { + if t == nil { + return nil, fmt.Errorf("translation: nil translation at index %d", i) + } + + from, err := canonical(t.from) + if err != nil { + return nil, err + } + + to, err := canonical(t.to) + if err != nil { + return nil, err + } + + if from == to { + return nil, fmt.Errorf("translation: %s translates onto itself", from) + } + + if _, dup := byMethod[from]; dup { + return nil, fmt.Errorf("translation: %s is translated twice", from) + } + + t.from, t.to = from, to + byMethod[from] = t + } + + return &Registry{byMethod: byMethod}, nil +} + +// Lookup returns the translation registered for fullMethod, reporting false when +// there is none and the call should be forwarded unchanged. A nil Registry, or +// one built from no translations, always reports false. +func (r *Registry) Lookup(fullMethod string) (*Translation, bool) { + if r == nil { + return nil, false + } + + t, ok := r.byMethod[fullMethod] + return t, ok +} + +// Methods returns the inbound methods the registry translates, in canonical +// "/pkg.Service/Method" form. Order is not significant. +func (r *Registry) Methods() []string { + if r == nil { + return nil + } + + out := make([]string, 0, len(r.byMethod)) + for method := range r.byMethod { + out = append(out, method) + } + + return out +} + +// stamp returns ctx with this translation's headers set on the outgoing +// metadata, or ctx unchanged when it has none. Set rather than append, so a +// value that arrived inbound and was forwarded cannot leave two on the wire. +func (t *Translation) stamp(ctx context.Context) context.Context { + if len(t.headers) == 0 { + return ctx + } + + return rpc.WithOutgoing(ctx, func(md metadata.MD) { + for key, value := range t.headers { + md.Set(key, value) + } + }) +} + +// From is the inbound method this translation replaces. +func (t *Translation) From() string { return t.from } + +// To is the upstream method that stands in for it. +func (t *Translation) To() string { return t.to } + +// canonical returns fullMethod in the leading-slash "/pkg.Service/Method" form +// gRPC hands an interceptor, so a mapping written either way still matches the +// method the interceptor is asked about. +func canonical(fullMethod string) (string, error) { + service, method, ok := rpc.ServiceMethod(fullMethod) + if !ok || service == "" || method == "" { + return "", fmt.Errorf("translation: %q is not a gRPC full method", fullMethod) + } + + return "/" + service + "/" + method, nil +} + +// newMessage allocates an empty T. The zero value of a generated message type is +// a nil pointer, which still carries its descriptor, so this works without the +// type registry the forwarder uses. +func newMessage[T proto.Message]() proto.Message { + var zero T + return zero.ProtoReflect().New().Interface() +} + +// nameOf returns the proto full name of T, so a type mismatch names the message +// a conversion expected rather than its Go type. +func nameOf[T proto.Message]() protoreflect.FullName { + var zero T + return zero.ProtoReflect().Descriptor().FullName() +} diff --git a/internal/cloud/translation/translation_test.go b/internal/cloud/translation/translation_test.go new file mode 100644 index 0000000..5f3bba1 --- /dev/null +++ b/internal/cloud/translation/translation_test.go @@ -0,0 +1,113 @@ +package translation_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + namespacepb "go.temporal.io/api/namespace/v1" + workflowservice "go.temporal.io/api/workflowservice/v1" + + "github.com/temporalio/temporal-proxy/internal/cloud/translation" +) + +// The mechanism is generic, so it is exercised with a made-up pairing rather than +// with a translation the proxy ships: DescribeNamespace onto GetSystemInfo. The +// types are unrelated in every way that matters here, which is the point - a +// test that used a real mapping's types could pass for the wrong reason. +const ( + fromMethod = "/temporal.api.workflowservice.v1.WorkflowService/DescribeNamespace" + toMethod = "/temporal.api.workflowservice.v1.WorkflowService/GetSystemInfo" +) + +func TestNewRegistryCanonicalizesMethods(t *testing.T) { + t.Parallel() + + // Written without the leading slash gRPC supplies, so the registry has to + // normalize both ends before it can match anything. + r, err := translation.NewRegistry( + translation.Adapt("pkg.Service/From", "pkg.Other/To", okRequest, okResponse), + ) + require.NoError(t, err) + + got, ok := r.Lookup("/pkg.Service/From") + require.True(t, ok) + require.Equal(t, "/pkg.Service/From", got.From()) + require.Equal(t, "/pkg.Other/To", got.To()) + require.Equal(t, []string{"/pkg.Service/From"}, r.Methods()) +} + +func TestNewRegistryRejectsBadMappings(t *testing.T) { + t.Parallel() + + cases := map[string]struct { + in []*translation.Translation + want string + }{ + "nil entry": { + in: []*translation.Translation{nil}, + want: "nil translation at index 0", + }, + "malformed from": { + in: []*translation.Translation{translation.Adapt("NotAMethod", toMethod, okRequest, okResponse)}, + want: `"NotAMethod" is not a gRPC full method`, + }, + "malformed to": { + in: []*translation.Translation{translation.Adapt(fromMethod, "NotAMethod", okRequest, okResponse)}, + want: `"NotAMethod" is not a gRPC full method`, + }, + "self mapping": { + in: []*translation.Translation{translation.Adapt(fromMethod, fromMethod, okRequest, okResponse)}, + want: "translates onto itself", + }, + "duplicate from": { + in: []*translation.Translation{ + translation.Adapt(fromMethod, toMethod, okRequest, okResponse), + translation.Adapt(fromMethod, "/pkg.Third/To", okRequest, okResponse), + }, + want: "is translated twice", + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + r, err := translation.NewRegistry(tc.in...) + require.ErrorContains(t, err, tc.want) + require.Nil(t, r) + }) + } +} + +func TestRegistryLookupMisses(t *testing.T) { + t.Parallel() + + var nilRegistry *translation.Registry + _, ok := nilRegistry.Lookup(fromMethod) + require.False(t, ok, "a nil registry translates nothing") + require.Nil(t, nilRegistry.Methods()) + + empty, err := translation.NewRegistry() + require.NoError(t, err) + _, ok = empty.Lookup(fromMethod) + require.False(t, ok) + require.Empty(t, empty.Methods()) +} + +// okRequest converts the caller's request into the substituted one. It carries +// nothing across, since what the conversions do is not what these tests are +// about. +func okRequest(*workflowservice.DescribeNamespaceRequest) (*workflowservice.GetSystemInfoRequest, error) { + return &workflowservice.GetSystemInfoRequest{}, nil +} + +// okResponse folds both the original request and the upstream reply into the +// caller's, so a test can tell which of them a value came from. +func okResponse( + req *workflowservice.DescribeNamespaceRequest, + up *workflowservice.GetSystemInfoResponse, + reply *workflowservice.DescribeNamespaceResponse, +) error { + reply.NamespaceInfo = &namespacepb.NamespaceInfo{Name: req.GetNamespace() + "@" + up.GetServerVersion()} + return nil +}