From 716c1e4b1888ccbae47e643d23ad4710e8860bce 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/translation/doc.go | 20 ++ internal/translation/interceptor.go | 115 ++++++++++ internal/translation/interceptor_test.go | 257 +++++++++++++++++++++++ internal/translation/translation.go | 213 +++++++++++++++++++ internal/translation/translation_test.go | 208 ++++++++++++++++++ 5 files changed, 813 insertions(+) create mode 100644 internal/translation/doc.go create mode 100644 internal/translation/interceptor.go create mode 100644 internal/translation/interceptor_test.go create mode 100644 internal/translation/translation.go create mode 100644 internal/translation/translation_test.go diff --git a/internal/translation/doc.go b/internal/translation/doc.go new file mode 100644 index 0000000..b6ce9e5 --- /dev/null +++ b/internal/translation/doc.go @@ -0,0 +1,20 @@ +// 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 [UnaryClientInterceptor] applies it: +// 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. +// +// This is distinct from the namespace translation in internal/proxy, which +// rewrites names inside a message but leaves the method alone. The two compose: +// install method translation as the innermost interceptor so namespace +// translation, payload codecs, and the reflective forwarder all keep seeing the +// method and message types the caller actually asked for. +// +// The package is the mechanism only; it ships no translations of its own. A +// [Translation] names the two methods and carries the conversions between their +// message types, so what is translated lives with the domain that needs it. +package translation diff --git a/internal/translation/interceptor.go b/internal/translation/interceptor.go new file mode 100644 index 0000000..e11a362 --- /dev/null +++ b/internal/translation/interceptor.go @@ -0,0 +1,115 @@ +package translation + +import ( + "context" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// Option configures the interceptor built by [UnaryClientInterceptor]. +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 frontend 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. +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 is Internal: the +// mapping is compiled in, so a failure there is a proxy bug and not something +// 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 status.Error(codes.Internal, err.Error()) + } + + 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 status.Error(codes.Internal, err.Error()) + } + + 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/translation/interceptor_test.go b/internal/translation/interceptor_test.go new file mode 100644 index 0000000..1807c2b --- /dev/null +++ b/internal/translation/interceptor_test.go @@ -0,0 +1,257 @@ +package translation + +import ( + "context" + "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" +) + +// systemInfoService is a fake upstream serving the method the test translation +// substitutes onto, recording the metadata it was called with. +type systemInfoService struct { + workflowservice.UnimplementedWorkflowServiceServer + + lis net.Listener + version string + + // mu guards md, which the serving goroutine writes while the test reads. + mu sync.Mutex + md metadata.MD +} + +func testRegistry(t *testing.T, opts ...func(*Translation)) *Registry { + t.Helper() + + tr := Adapt(fromMethod, toMethod, okRequest, okResponse) + for _, opt := range opts { + opt(tr) + } + + r, err := NewRegistry(tr) + require.NoError(t, err) + + return r +} + +func TestUnaryClientInterceptorPassesUntranslatedMethodsThrough(t *testing.T) { + t.Parallel() + + var gotMethod string + invoker := func(_ context.Context, method string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + gotMethod = method + return nil + } + + const other = "/temporal.api.workflowservice.v1.WorkflowService/ListNamespaces" + err := UnaryClientInterceptor(testRegistry(t))( + t.Context(), other, + &workflowservice.ListNamespacesRequest{}, &workflowservice.ListNamespacesResponse{}, nil, invoker, + ) + require.NoError(t, err) + require.Equal(t, other, gotMethod) +} + +func TestUnaryClientInterceptorSubstitutesTheUpstreamCall(t *testing.T) { + t.Parallel() + + var ( + gotMethod string + gotReq *workflowservice.GetSystemInfoRequest + ) + + invoker := func(_ context.Context, method string, req, reply any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + gotMethod = method + gotReq = mustProto[*workflowservice.GetSystemInfoRequest](t, req) + mustProto[*workflowservice.GetSystemInfoResponse](t, reply).ServerVersion = "1.2.3" + + return nil + } + + reply := &workflowservice.DescribeNamespaceResponse{} + err := UnaryClientInterceptor(testRegistry(t))( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply, nil, invoker, + ) + require.NoError(t, err) + + require.Equal(t, toMethod, gotMethod, "the upstream sees the substituted method") + require.NotNil(t, gotReq, "and the converted request") + require.Equal(t, "payments@1.2.3", reply.GetNamespaceInfo().GetName()) +} + +func TestUnaryClientInterceptorReturnsTheUpstreamError(t *testing.T) { + t.Parallel() + + want := status.Error(codes.PermissionDenied, "no") + invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { + return want + } + + err := UnaryClientInterceptor(testRegistry(t))( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker, + ) + require.Equal(t, want, err, "the caller sees the upstream's status, untranslated") +} + +func TestUnaryClientInterceptorReportsConversionFailureAsInternal(t *testing.T) { + t.Parallel() + + // A reply of the wrong type for the registered translation: the request + // converts, the call succeeds, and folding the reply back fails. + invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { + return nil + } + + err := UnaryClientInterceptor(testRegistry(t))( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.ListNamespacesResponse{}, nil, invoker, + ) + require.Equal(t, codes.Internal, status.Code(err)) +} + +func TestUnaryClientInterceptorForwardsNonProtoCallsUnchanged(t *testing.T) { + t.Parallel() + + var gotMethod string + invoker := func(_ context.Context, method string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + gotMethod = method + return nil + } + + err := UnaryClientInterceptor(testRegistry(t))(t.Context(), fromMethod, "req", "reply", nil, invoker) + require.NoError(t, err) + require.Equal(t, fromMethod, gotMethod, "nothing to convert, so nothing is substituted") +} + +func TestUnaryClientInterceptorStampsHeadersOnTheSubstitutedCallOnly(t *testing.T) { + t.Parallel() + + r := testRegistry(t, func(tr *Translation) { tr.WithHeader("x-api-version", "v1") }) + + var gotMD metadata.MD + invoker := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + gotMD, _ = metadata.FromOutgoingContext(ctx) + return nil + } + + err := UnaryClientInterceptor(r)( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker, + ) + require.NoError(t, err) + require.Equal(t, []string{"v1"}, gotMD.Get("x-api-version")) + + // An untranslated call keeps whatever the caller sent. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs("x-api-version", "caller")) + err = UnaryClientInterceptor(r)( + ctx, "/pkg.Other/Untranslated", + &workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker, + ) + require.NoError(t, err) + require.Equal(t, []string{"caller"}, gotMD.Get("x-api-version")) +} + +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. + elsewhere := newSystemInfoService(t, "9.9.9") + + cc, err := grpc.NewClient(elsewhere.addr(t), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + var chained bool + invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { + chained = true + return nil + } + + reply := &workflowservice.DescribeNamespaceResponse{} + err = UnaryClientInterceptor(testRegistry(t), Via(cc))( + t.Context(), fromMethod, + &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply, nil, invoker, + ) + require.NoError(t, err) + + require.False(t, chained, "the call must leave the chain rather than continue down it") + require.Equal(t, "payments@9.9.9", reply.GetNamespaceInfo().GetName()) +} + +func TestDialOptionsTranslateOverARealConnection(t *testing.T) { + t.Parallel() + + upstream := newSystemInfoService(t, "1.2.3") + + dialOpts := append( + []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, + DialOptions(testRegistry(t, func(tr *Translation) { tr.WithHeader("x-api-version", "v1") }))..., + ) + cc, err := grpc.NewClient(upstream.addr(t), dialOpts...) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + // Invoke the inbound method by name, the way the reflective forwarder does: + // the interceptor is what turns it into the substituted call. + 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.Equal(t, []string{"v1"}, upstream.metadata().Get("x-api-version"), "the header reached the wire") +} + +func (s *systemInfoService) GetSystemInfo( + ctx context.Context, _ *workflowservice.GetSystemInfoRequest, +) (*workflowservice.GetSystemInfoResponse, error) { + md, _ := metadata.FromIncomingContext(ctx) + + s.mu.Lock() + defer s.mu.Unlock() + + s.md = md + + return &workflowservice.GetSystemInfoResponse{ServerVersion: s.version}, nil +} + +// newSystemInfoService starts the fake on a loopback port and stops it when the +// test ends. +func newSystemInfoService(t *testing.T, version string) *systemInfoService { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + svc := &systemInfoService{lis: lis, version: version} + + svr := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(svr, svc) + go func() { _ = svr.Serve(lis) }() + t.Cleanup(svr.Stop) + + return svc +} + +func (s *systemInfoService) addr(t *testing.T) string { + t.Helper() + return s.lis.Addr().String() +} + +func (s *systemInfoService) metadata() metadata.MD { + s.mu.Lock() + defer s.mu.Unlock() + + return s.md.Copy() +} diff --git a/internal/translation/translation.go b/internal/translation/translation.go new file mode 100644 index 0000000..f51fb1a --- /dev/null +++ b/internal/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/translation/translation_test.go b/internal/translation/translation_test.go new file mode 100644 index 0000000..a2e002f --- /dev/null +++ b/internal/translation/translation_test.go @@ -0,0 +1,208 @@ +package translation + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + namespacepb "go.temporal.io/api/namespace/v1" + workflowservice "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc/metadata" + "google.golang.org/protobuf/proto" +) + +// 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 := NewRegistry(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 + want string + }{ + "nil entry": { + in: []*Translation{nil}, + want: "nil translation at index 0", + }, + "malformed from": { + in: []*Translation{Adapt("NotAMethod", toMethod, okRequest, okResponse)}, + want: `"NotAMethod" is not a gRPC full method`, + }, + "malformed to": { + in: []*Translation{Adapt(fromMethod, "NotAMethod", okRequest, okResponse)}, + want: `"NotAMethod" is not a gRPC full method`, + }, + "self mapping": { + in: []*Translation{Adapt(fromMethod, fromMethod, okRequest, okResponse)}, + want: "translates onto itself", + }, + "duplicate from": { + in: []*Translation{ + Adapt(fromMethod, toMethod, okRequest, okResponse), + 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 := NewRegistry(tc.in...) + require.ErrorContains(t, err, tc.want) + require.Nil(t, r) + }) + } +} + +func TestRegistryLookupMisses(t *testing.T) { + t.Parallel() + + var nilRegistry *Registry + _, ok := nilRegistry.Lookup(fromMethod) + require.False(t, ok, "a nil registry translates nothing") + require.Nil(t, nilRegistry.Methods()) + + empty, err := NewRegistry() + require.NoError(t, err) + _, ok = empty.Lookup(fromMethod) + require.False(t, ok) + require.Empty(t, empty.Methods()) +} + +func TestAdaptRejectsMismatchedMessages(t *testing.T) { + t.Parallel() + + tr := Adapt(fromMethod, toMethod, okRequest, okResponse) + + _, err := tr.request(&workflowservice.ListNamespacesRequest{}) + require.ErrorContains(t, err, "wanted a temporal.api.workflowservice.v1.DescribeNamespaceRequest request") + + req := &workflowservice.DescribeNamespaceRequest{} + err = tr.response(req, &workflowservice.ListNamespacesResponse{}, &workflowservice.DescribeNamespaceResponse{}) + require.ErrorContains(t, err, "wanted a temporal.api.workflowservice.v1.GetSystemInfoResponse reply") + + err = tr.response(req, &workflowservice.GetSystemInfoResponse{}, &workflowservice.ListNamespacesResponse{}) + require.ErrorContains(t, err, "wanted a temporal.api.workflowservice.v1.DescribeNamespaceResponse reply") +} + +func TestAdaptPropagatesConversionErrors(t *testing.T) { + t.Parallel() + + boom := errors.New("boom") + tr := Adapt( + fromMethod, + toMethod, + func(*workflowservice.DescribeNamespaceRequest) (*workflowservice.GetSystemInfoRequest, error) { + return nil, boom + }, + okResponse, + ) + + _, err := tr.request(&workflowservice.DescribeNamespaceRequest{}) + require.ErrorIs(t, err, boom) +} + +func TestAdaptGivesTheResponseConverterTheOriginalRequest(t *testing.T) { + t.Parallel() + + // A field the upstream request cannot carry can only be honoured on the way + // back, which is why the response converter is handed the request too. + tr := Adapt(fromMethod, toMethod, okRequest, okResponse) + + reply := &workflowservice.DescribeNamespaceResponse{} + err := tr.response( + &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, + &workflowservice.GetSystemInfoResponse{ServerVersion: "1.2.3"}, + reply, + ) + require.NoError(t, err) + require.Equal(t, "payments@1.2.3", reply.GetNamespaceInfo().GetName()) +} + +func TestTranslationAllocatesUpstreamReply(t *testing.T) { + t.Parallel() + + tr := Adapt(fromMethod, toMethod, okRequest, okResponse) + + reply := tr.reply() + require.IsType(t, &workflowservice.GetSystemInfoResponse{}, reply) + require.NotSame(t, reply, tr.reply(), "each call allocates its own reply") +} + +func TestWithHeaderStampsOnlyWhenDeclared(t *testing.T) { + t.Parallel() + + tr := Adapt(fromMethod, toMethod, okRequest, okResponse).WithHeader("x-api-version", "v1") + + md, ok := metadata.FromOutgoingContext(tr.stamp(t.Context())) + require.True(t, ok) + require.Equal(t, []string{"v1"}, md.Get("x-api-version")) + + // A translation that declares no header leaves the context alone. + plain := Adapt(fromMethod, toMethod, okRequest, okResponse) + ctx := t.Context() + require.Equal(t, ctx, plain.stamp(ctx)) +} + +func TestWithHeaderReplacesAnInboundValue(t *testing.T) { + t.Parallel() + + tr := Adapt(fromMethod, toMethod, okRequest, okResponse).WithHeader("x-api-version", "v2") + + // The caller's own value was forwarded onto the outgoing context. 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("x-api-version", "v1")) + + md, ok := metadata.FromOutgoingContext(tr.stamp(ctx)) + require.True(t, ok) + require.Equal(t, []string{"v2"}, md.Get("x-api-version")) +} + +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 +} + +// mustProto fails the test unless m is the message the caller expected. +func mustProto[T proto.Message](t *testing.T, m any) T { + t.Helper() + + out, ok := m.(T) + require.True(t, ok, "got %T", m) + return out +}