fix(hop): one service's gRPC and Connect endpoints are one destination

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.

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
 	}

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)
+		}
+	}
+}

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())
+	}
+}