From 4284327ab832e11c7511d5c2f1baa9f270194930 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:41:26 -0400 Subject: [PATCH] fix(hop): one service's gRPC and Connect endpoints are one destination Since #26 a Connect endpoint also claims its procedures for gRPC, so a module that still publishes the same service on a grpc and a connect endpoint (annotations, model, robin, tracing) made the hop refuse its own routes at startup: "ambiguous destination for annotations /annotations.v1.AnnotationsService/Annotate". Routing now treats that pair as one destination when both endpoints carry the same contract (package, version, digest): gRPC calls go to the Connect endpoint, which serves gRPC, gRPC-Web and Connect; the gRPC endpoint keeps only procedures Connect does not publish, and a gRPC-only module routes as before. Every other second claim is still refused, now naming both listeners: two services, two endpoints of one transport, an HTTP route over a Connect procedure, or one service's endpoints under different contracts. Resolve no longer returns the first exact match of an unvalidated snapshot; it applies the same rule. Co-Authored-By: Claude Opus 5.5 --- libraries/hop/go/routing/README.md | 15 ++-- libraries/hop/go/routing/routes.go | 58 ++++++++++++--- libraries/hop/go/routing/routes_test.go | 73 +++++++++++++++++++ .../pkg/infra/proxy/connect_listener_test.go | 41 +++++++++++ 4 files changed, 171 insertions(+), 16 deletions(-) diff --git a/libraries/hop/go/routing/README.md b/libraries/hop/go/routing/README.md index 4a60fde3..4c7ff6fe 100644 --- a/libraries/hop/go/routing/README.md +++ b/libraries/hop/go/routing/README.md @@ -13,8 +13,13 @@ must be origins; they cannot add a path prefix after the authority check. A declared Connect endpoint serves both Connect HTTP requests and gRPC calls at -its one resolved address. Both use the declared protobuf method's policy. A gRPC -and Connect publication for the same audience/method is ambiguous and refused; -remove the separate gRPC endpoint when adopting one Connect listener. Unknown -methods, undeclared audiences, unsafe destinations and transport refusals remain -closed. The hop dials using the shared host transport policy. +its one resolved address. Both use the declared protobuf method's policy. When +one service publishes the same contract (package, version and digest) on both +its gRPC endpoint and its Connect endpoint, that is one destination: gRPC calls +for a procedure on both go to the Connect endpoint, and the gRPC endpoint serves +only procedures the Connect endpoint does not publish. Any other second claim on +a procedure is ambiguous and refused, naming both listeners: two services, two +endpoints of one transport, an HTTP route over a Connect procedure, or one +service's two endpoints under different contracts. Unknown methods, undeclared +audiences, unsafe destinations and transport refusals remain closed. The hop +dials using the shared host transport policy. diff --git a/libraries/hop/go/routing/routes.go b/libraries/hop/go/routing/routes.go index 7a5e4d6c..c6be83fd 100644 --- a/libraries/hop/go/routing/routes.go +++ b/libraries/hop/go/routing/routes.go @@ -60,7 +60,7 @@ func (r *Routing) Validate() error { if r.Schema != RoutingSchema || len(r.Endpoints) == 0 { return fmt.Errorf("hop routes: expected %s and at least one endpoint", RoutingSchema) } - seen, listeners := map[string]bool{}, map[string]bool{} + seen, listeners := map[string]RouteEndpoint{}, map[string]bool{} for _, e := range r.Endpoints { if e.Audience == "" || e.Service == "" || e.Endpoint == "" || e.Address == "" || e.Package == "" || e.Version == "" || e.Digest == "" || len(e.Methods) == 0 { return fmt.Errorf("hop routes: incomplete published endpoint identity") @@ -93,23 +93,43 @@ func (r *Routing) Validate() error { if transport == "connect" { transport = "http" } - key := e.Audience + "\x00" + transport + "\x00" + match - if seen[key] { - return fmt.Errorf("hop routes: ambiguous destination for %s %s", e.Audience, method) - } - seen[key] = true + keys := []string{e.Audience + "\x00" + transport + "\x00" + match} if e.Transport == "connect" { - grpcKey := e.Audience + "\x00grpc\x00" + method - if seen[grpcKey] { - return fmt.Errorf("hop routes: ambiguous destination for %s %s", e.Audience, method) + keys = append(keys, e.Audience+"\x00grpc\x00"+method) + } + for _, key := range keys { + if prior, taken := seen[key]; taken { + if err := sameDestination(prior, e); err != nil { + return fmt.Errorf("hop routes: ambiguous destination for %s %s: %w", e.Audience, method, err) + } + continue } - seen[grpcKey] = true + seen[key] = e } } } return nil } +// sameDestination accepts one collision only: a service publishing the same +// contract on its gRPC endpoint and on its Connect endpoint, which also serves +// gRPC. That is one destination with one policy; Resolve prefers Connect. +// Anything else claiming a procedure twice is refused by name. +func sameDestination(a, b RouteEndpoint) error { + name := func(e RouteEndpoint) string { + return e.Audience + "/" + e.Service + "/" + e.Endpoint + " (" + e.Transport + ")" + } + if a.Audience != b.Audience || a.Service != b.Service || a.Transport == b.Transport || + (a.Transport != "grpc" && a.Transport != "connect") || (b.Transport != "grpc" && b.Transport != "connect") { + return fmt.Errorf("published by %s and %s", name(a), name(b)) + } + if a.Package != b.Package || a.Version != b.Version || a.Digest != b.Digest { + return fmt.Errorf("%s and %s publish it under different contracts (%s@%s %s, %s@%s %s), so their policies may differ", + name(a), name(b), a.Package, a.Version, a.Digest, b.Package, b.Version, b.Digest) + } + return nil +} + // Find selects a published route after the caller's audience has been checked. // HTTP includes Connect POSTs, selected from publication, never path heuristics. // Exact matches win, as in method admission; overlapping templates are refused. @@ -140,6 +160,7 @@ func (r *Routing) Resolve(audience, transport, method string) (RouteEndpoint, st policy string } var matched []match + var exact *match for _, endpoint := range r.Endpoints { if endpoint.Audience != audience || (endpoint.Transport != transport && !((transport == "http" || transport == "grpc") && endpoint.Transport == "connect")) { continue @@ -151,7 +172,19 @@ func (r *Routing) Resolve(audience, transport, method string) (RouteEndpoint, st declared = "POST " + declared } if declared == method { - return endpoint, policy, nil + if exact == nil { + exact = &match{endpoint, policy} + continue + } + // Only one service's gRPC and Connect endpoints may share a + // procedure; Connect serves gRPC too, so it is the destination. + if err := sameDestination(exact.endpoint, endpoint); err != nil { + return RouteEndpoint{}, "", fmt.Errorf("hop routes: ambiguous destination for %s %s: %w", audience, method, err) + } + if endpoint.Transport == "connect" { + exact = &match{endpoint, policy} + } + continue } verb, path, ok := strings.Cut(method, " ") wantVerb, template, wantOK := strings.Cut(declared, " ") @@ -160,6 +193,9 @@ func (r *Routing) Resolve(audience, transport, method string) (RouteEndpoint, st } } } + if exact != nil { + return exact.endpoint, exact.policy, nil + } if len(matched) == 1 { return matched[0].endpoint, matched[0].policy, nil } diff --git a/libraries/hop/go/routing/routes_test.go b/libraries/hop/go/routing/routes_test.go index 26943b68..60721a51 100644 --- a/libraries/hop/go/routing/routes_test.go +++ b/libraries/hop/go/routing/routes_test.go @@ -125,3 +125,76 @@ func TestConnectGRPCOverlapRefusesInEitherOrder(t *testing.T) { } } } + +// A module that still publishes one service on a gRPC and a Connect endpoint +// (the same contract) is one destination: gRPC calls go to Connect, which +// serves gRPC, gRPC-Web and Connect, in either publication order. +func TestOneServiceOnGRPCAndConnectIsOneDestination(t *testing.T) { + grpcFace, connectFace := endpoint("annotations", "grpc", "/a.Service/Call", "/a.Service/GRPCOnly"), endpoint("annotations", "connect", "/a.Service/Call") + for _, endpoints := range [][]RouteEndpoint{{grpcFace, connectFace}, {connectFace, grpcFace}} { + r := Routing{Schema: RoutingSchema, Endpoints: endpoints} + if err := r.Validate(); err != nil { + t.Fatal(err) + } + for _, tc := range []struct{ transport, method, endpoint, policy string }{ + {"grpc", "/a.Service/Call", "connect", "/a.Service/Call"}, + {"http", "POST /a.Service/Call", "connect", "/a.Service/Call"}, + {"grpc", "/a.Service/GRPCOnly", "grpc", "/a.Service/GRPCOnly"}, + } { + got, policy, err := r.Resolve("module", tc.transport, tc.method) + if err != nil || got.Endpoint != tc.endpoint || policy != tc.policy { + t.Fatalf("%s %s: %+v %q %v", tc.transport, tc.method, got, policy, err) + } + } + } +} + +func TestGRPCOnlyModuleStillRoutes(t *testing.T) { + r := Routing{Schema: RoutingSchema, Endpoints: []RouteEndpoint{endpoint("legacy", "grpc", "/legacy.Service/Call")}} + if err := r.Validate(); err != nil { + t.Fatal(err) + } + got, policy, err := r.Resolve("module", "grpc", "/legacy.Service/Call") + if err != nil || got.Transport != "grpc" || policy != "/legacy.Service/Call" { + t.Fatalf("%+v %q %v", got, policy, err) + } + if _, _, err := r.Resolve("module", "http", "POST /legacy.Service/Call"); err == nil { + t.Fatal("a gRPC listener answered a Connect request") + } +} + +func TestTwoServicesClaimingOneProcedureRefuseByName(t *testing.T) { + for _, pair := range [][2]RouteEndpoint{ + {endpoint("first", "connect", "/a.Service/Call"), endpoint("second", "grpc", "/a.Service/Call")}, + {endpoint("first", "connect", "/a.Service/Call"), endpoint("second", "connect", "/a.Service/Call")}, + {endpoint("first", "grpc", "/a.Service/Call"), endpoint("second", "grpc", "/a.Service/Call")}, + } { + r := Routing{Schema: RoutingSchema, Endpoints: pair[:]} + err := r.Validate() + if err == nil || !strings.Contains(err.Error(), "module/first/") || !strings.Contains(err.Error(), "module/second/") { + t.Fatalf("expected a refusal naming both services, got %v", err) + } + // Unvalidated snapshots refuse at resolution too, never first-wins. + if got, _, err := r.Resolve("module", "grpc", "/a.Service/Call"); err == nil { + t.Fatalf("ambiguous procedure selected %+v", got) + } + } +} + +func TestSameServiceWithDifferentContractsRefuses(t *testing.T) { + for _, change := range []func(*RouteEndpoint){ + func(e *RouteEndpoint) { e.Digest = "sha256:other" }, + func(e *RouteEndpoint) { e.Version = "v2" }, + func(e *RouteEndpoint) { e.Package = "owner/other" }, + } { + connectFace := endpoint("annotations", "connect", "/a.Service/Call") + change(&connectFace) + r := Routing{Schema: RoutingSchema, Endpoints: []RouteEndpoint{endpoint("annotations", "grpc", "/a.Service/Call"), connectFace}} + if err := r.Validate(); err == nil || !strings.Contains(err.Error(), "different contracts") { + t.Fatalf("differing policies accepted: %v", err) + } + if got, _, err := r.Resolve("module", "grpc", "/a.Service/Call"); err == nil { + t.Fatalf("differing policies selected %+v", got) + } + } +} diff --git a/modules/authority/services/hop/code/pkg/infra/proxy/connect_listener_test.go b/modules/authority/services/hop/code/pkg/infra/proxy/connect_listener_test.go index 7b548033..0601276c 100644 --- a/modules/authority/services/hop/code/pkg/infra/proxy/connect_listener_test.go +++ b/modules/authority/services/hop/code/pkg/infra/proxy/connect_listener_test.go @@ -83,3 +83,44 @@ func TestGRPCAndConnectReachOnePublishedConnectListener(t *testing.T) { t.Fatalf("refusal reached backend: %d", m.calls.Load()) } } + +// A module that still publishes its service on a gRPC endpoint beside its +// Connect endpoint is one destination: the hop starts and sends gRPC to Connect. +func TestDualPublishedServiceSendsGRPCToItsConnectListener(t *testing.T) { + t.Setenv("CODEFLY__ENVIRONMENT", "local") + dial := func(m *connectModule) (string, *grpc.ClientConn) { + _, handler := genconnect.NewHopServiceHandler(m) + backend := httptest.NewServer(h2c.NewHandler(handler, &http2.Server{})) + t.Cleanup(backend.Close) + target, options, err := host.Endpoint(t.Context(), backend.URL) + if err != nil { + t.Fatal(err) + } + conn, err := grpc.NewClient(target, append(host.DialOptions(options), grpc.WithDefaultCallOptions(proxy.CallOptions()...))...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return backend.URL, conn + } + connectFace, grpcFace := &connectModule{}, &connectModule{} + connectURL, connectConn := dial(connectFace) + grpcURL, grpcConn := dial(grpcFace) + routes := &routing.Routing{Schema: routing.RoutingSchema, Endpoints: []routing.RouteEndpoint{ + routeEndpoint("module", "grpc", grpcURL, genconnect.HopServiceVersionProcedure), + routeEndpoint("module", "connect", connectURL, genconnect.HopServiceVersionProcedure), + }} + if err := routes.Validate(); err != nil { + t.Fatal(err) + } + decider := decide.Decider{Routes: routes, Authority: &routeChecker{}, Modules: map[string]bool{"module": true}, AudienceOf: func(string) (string, error) { return "module", nil }} + p := &proxy.Proxy{Decide: decider.Decide, Proof: proof, Upstreams: map[string]*grpc.ClientConn{connectURL: connectConn, grpcURL: grpcConn}} + conn := serve(t, p.ServerOptions(), func(*grpc.Server) {}) + ctx := metadata.AppendToOutgoingContext(t.Context(), hop.WorkContextKey, "context") + if _, err := gen.NewHopServiceClient(conn).Version(ctx, &gen.VersionRequest{}); err != nil { + t.Fatal(err) + } + if connectFace.calls.Load() != 1 || grpcFace.calls.Load() != 0 { + t.Fatalf("connect=%d grpc=%d", connectFace.calls.Load(), grpcFace.calls.Load()) + } +}