fix(vhost): isolate connections across HTTP route registrations

pkg/util/vhost/http.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"strconv"
 	"time"
 
 	libio "github.com/fatedier/golib/io"
@@ -78,11 +79,12 @@ func NewHTTPReverseProxy(option HTTPReverseProxyOptions, vhostRouter *Routers) *
 					log.Tracef("choose endpoint name [%s] for http request host [%s] path [%s] httpuser [%s]",
 						endpoint, originalHost, reqRouteInfo.URL, reqRouteInfo.HTTPUser)
 				}
-				// Set {domain}.{location}.{routeByHTTPUser}.{endpoint} as URL host here to let http transport reuse connections.
+				// Keep connection reuse within the same route registration and endpoint.
 				req.URL.Host = rc.Domain + "." +
 					base64.StdEncoding.EncodeToString([]byte(rc.Location)) + "." +
 					base64.StdEncoding.EncodeToString([]byte(rc.RouteByHTTPUser)) + "." +
-					base64.StdEncoding.EncodeToString([]byte(endpoint))
+					base64.StdEncoding.EncodeToString([]byte(endpoint)) + "." +
+					strconv.FormatUint(reqRouteInfo.routeID, 10)
 
 				for k, v := range rc.Headers {
 					req.Header.Set(k, v)
@@ -106,7 +108,7 @@ func NewHTTPReverseProxy(option HTTPReverseProxyOptions, vhostRouter *Routers) *
 			IdleConnTimeout:       60 * time.Second,
 			MaxIdleConnsPerHost:   5,
 			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
-				return rp.CreateConnection(ctx.Value(RouteInfoKey).(*RequestRouteInfo), true)
+				return createConnection(ctx.Value(RouteInfoKey).(*RequestRouteInfo), ctx.Value(RouteConfigKey).(*RouteConfig), true)
 			},
 			Proxy: func(req *http.Request) (*url.URL, error) {
 				// Use proxy mode if there is host in HTTP first request line.
@@ -156,31 +158,21 @@ func (rp *HTTPReverseProxy) UnRegister(routeCfg RouteConfig) {
 	rp.vhostRouter.Del(routeCfg.Domain, routeCfg.Location, routeCfg.RouteByHTTPUser)
 }
 
-func (rp *HTTPReverseProxy) GetRouteConfig(domain, location, routeByHTTPUser string) *RouteConfig {
-	vr, ok := rp.vhostRouter.getByRoute(domain, location, routeByHTTPUser)
-	if ok {
-		log.Debugf("get new http request host [%s] path [%s] httpuser [%s]", domain, location, routeByHTTPUser)
-		return vr.payload.(*RouteConfig)
-	}
-	return nil
-}
-
-// CreateConnection create a new connection by route config
-func (rp *HTTPReverseProxy) CreateConnection(reqRouteInfo *RequestRouteInfo, byEndpoint bool) (net.Conn, error) {
-	host, _ := httppkg.CanonicalHost(reqRouteInfo.Host)
-	vr, ok := rp.vhostRouter.getByRoute(host, reqRouteInfo.URL, reqRouteInfo.HTTPUser)
-	if ok {
+// createConnection uses the same route configuration that authorized the request.
+func createConnection(reqRouteInfo *RequestRouteInfo, rc *RouteConfig, byEndpoint bool) (net.Conn, error) {
+	if rc != nil {
 		if byEndpoint {
-			fn := vr.payload.(*RouteConfig).CreateConnByEndpointFn
+			fn := rc.CreateConnByEndpointFn
 			if fn != nil {
 				return fn(reqRouteInfo.Endpoint, reqRouteInfo.RemoteAddr)
 			}
 		}
-		fn := vr.payload.(*RouteConfig).CreateConnFn
+		fn := rc.CreateConnFn
 		if fn != nil {
 			return fn(reqRouteInfo.RemoteAddr)
 		}
 	}
+	host, _ := httppkg.CanonicalHost(reqRouteInfo.Host)
 	return nil, fmt.Errorf("%v: %s %s %s", ErrNoRouteFound, host, reqRouteInfo.URL, reqRouteInfo.HTTPUser)
 }
 
@@ -234,7 +226,8 @@ func (rp *HTTPReverseProxy) connectHandler(rw http.ResponseWriter, req *http.Req
 		return
 	}
 
-	remote, err := rp.CreateConnection(req.Context().Value(RouteInfoKey).(*RequestRouteInfo), false)
+	remote, err := createConnection(
+		req.Context().Value(RouteInfoKey).(*RequestRouteInfo), req.Context().Value(RouteConfigKey).(*RouteConfig), false)
 	if err != nil {
 		_ = NotFoundResponse().Write(client)
 		client.Close()
@@ -277,7 +270,12 @@ func (rp *HTTPReverseProxy) injectRequestInfoToCtx(req *http.Request) *http.Requ
 	}
 
 	originalHost, _ := httppkg.CanonicalHost(reqRouteInfo.Host)
-	rc := rp.GetRouteConfig(originalHost, reqRouteInfo.URL, reqRouteInfo.HTTPUser)
+	var rc *RouteConfig
+	if vr, ok := rp.vhostRouter.getByRoute(originalHost, reqRouteInfo.URL, reqRouteInfo.HTTPUser); ok {
+		rc = vr.payload.(*RouteConfig)
+		reqRouteInfo.routeID = vr.id
+		log.Debugf("get new http request host [%s] path [%s] httpuser [%s]", originalHost, reqRouteInfo.URL, reqRouteInfo.HTTPUser)
+	}
 
 	newctx := req.Context()
 	newctx = context.WithValue(newctx, RouteInfoKey, reqRouteInfo)

pkg/util/vhost/http_route_reuse_test.go

@@ -0,0 +1,142 @@
+package vhost
+
+import (
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHTTPRouteReplacementUsesNewBackend(t *testing.T) {
+	oldBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "old backend")
+	}))
+	t.Cleanup(oldBackend.Close)
+	newBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "new backend")
+	}))
+	t.Cleanup(newBackend.Close)
+
+	var oldDials, newDials atomic.Int32
+	oldRoute := RouteConfig{
+		Domain: "example.com",
+		CreateConnFn: func(string) (net.Conn, error) {
+			oldDials.Add(1)
+			return net.Dial("tcp", oldBackend.Listener.Addr().String())
+		},
+	}
+	newRoute := RouteConfig{
+		Domain: oldRoute.Domain,
+		CreateConnFn: func(string) (net.Conn, error) {
+			newDials.Add(1)
+			return net.Dial("tcp", newBackend.Listener.Addr().String())
+		},
+	}
+	rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
+	require.NoError(t, rp.Register(oldRoute))
+	frontend := httptest.NewServer(rp)
+	t.Cleanup(frontend.Close)
+
+	client := &http.Client{
+		// Each request opens a fresh frontend connection, so only the proxy's
+		// backend connection pool can retain the old destination.
+		Transport: &http.Transport{DisableKeepAlives: true},
+		Timeout:   5 * time.Second,
+	}
+	t.Cleanup(client.CloseIdleConnections)
+	request := func() string {
+		t.Helper()
+		req, err := http.NewRequest(http.MethodGet, frontend.URL, nil)
+		require.NoError(t, err)
+		req.Host = oldRoute.Domain
+		resp, err := client.Do(req)
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		require.Equal(t, http.StatusOK, resp.StatusCode)
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		return string(body)
+	}
+
+	require.Equal(t, "old backend", request())
+	require.Equal(t, "old backend", request())
+	require.EqualValues(t, 1, oldDials.Load(), "the unchanged route should reuse its backend connection")
+
+	rp.UnRegister(oldRoute)
+	require.NoError(t, rp.Register(newRoute))
+
+	require.Equal(t, "new backend", request(), "a new request must use the replacement route")
+	require.EqualValues(t, 1, newDials.Load())
+}
+
+func TestHTTPRouteReplacementKeepsMatchedConfiguration(t *testing.T) {
+	oldBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "old backend")
+	}))
+	t.Cleanup(oldBackend.Close)
+	newBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "new backend")
+	}))
+	t.Cleanup(newBackend.Close)
+
+	rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
+	oldRoute := RouteConfig{
+		Domain:   "example.com",
+		Username: "alice",
+		Password: "old password",
+		CreateConnFn: func(string) (net.Conn, error) {
+			return net.Dial("tcp", oldBackend.Listener.Addr().String())
+		},
+	}
+	newRoute := RouteConfig{
+		Domain:   oldRoute.Domain,
+		Username: "alice",
+		Password: "new password",
+		CreateConnFn: func(string) (net.Conn, error) {
+			return net.Dial("tcp", newBackend.Listener.Addr().String())
+		},
+	}
+	replaced := make(chan error, 1)
+	oldRoute.ChooseEndpointFn = func() (string, error) {
+		// Rewrite runs after route matching and authentication, but before
+		// Transport dials. Replace the route at this exact boundary.
+		rp.UnRegister(oldRoute)
+		replaced <- rp.Register(newRoute)
+		return "", nil
+	}
+	require.NoError(t, rp.Register(oldRoute))
+	frontend := httptest.NewServer(rp)
+	t.Cleanup(frontend.Close)
+	client := &http.Client{Timeout: 5 * time.Second}
+
+	request := func(password string) (int, string) {
+		t.Helper()
+		req, err := http.NewRequest(http.MethodGet, frontend.URL, nil)
+		require.NoError(t, err)
+		req.Host = oldRoute.Domain
+		req.SetBasicAuth("alice", password)
+		resp, err := client.Do(req)
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		return resp.StatusCode, string(body)
+	}
+
+	status, body := request("old password")
+	require.NoError(t, <-replaced)
+	require.Equal(t, http.StatusOK, status)
+	require.Equal(t, "old backend", body, "dial the route whose policy authorized this request")
+
+	status, body = request("new password")
+	require.Equal(t, http.StatusOK, status)
+	require.Equal(t, "new backend", body)
+
+	status, _ = request("old password")
+	require.Equal(t, http.StatusUnauthorized, status, "later requests must use the replacement policy")
+}

pkg/util/vhost/router.go

@@ -14,11 +14,13 @@ type routerByHTTPUser map[string][]*Router
 
 type Routers struct {
 	indexByDomain map[string]routerByHTTPUser
+	nextID        uint64
 
 	mutex sync.RWMutex
 }
 
 type Router struct {
+	id       uint64
 	domain   string
 	location string
 	httpUser string
@@ -52,7 +54,9 @@ func (r *Routers) Add(domain, location, httpUser string, payload any) error {
 		vrs = make([]*Router, 0, 1)
 	}
 
+	r.nextID++
 	vr := &Router{
+		id:       r.nextID,
 		domain:   domain,
 		location: location,
 		httpUser: httpUser,

pkg/util/vhost/vhost.go

@@ -39,6 +39,7 @@ type RequestRouteInfo struct {
 	RemoteAddr string
 	URLHost    string
 	Endpoint   string
+	routeID    uint64
 }
 
 type (