fix(proxy): reject cached sessions on draining CNs (#25913)

## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #25895

## What this PR does / why we need it:

Normal proxy routing obtains candidates from `MOCluster.GetCNService`,
which
excludes draining and removed CNs. Cached backend reuse bypassed that
candidate
selection and `router.CanReuseCachedCN` checked only the CN health
circuit
breaker.

A draining CN can remain breaker-healthy, so a fresh client login could
continue
to reuse a cached backend session on a CN that normal routing had
already made
ineligible. This could delay or defeat graceful draining and scale-in.

This change requires cached reuse to pass both existing gates:

- the health breaker must allow the CN; and
- the CN UUID must still be present in the current `GetCNService`
routable
  cluster view.

The existing nil-entry behavior is preserved. Missing and draining CNs
are
rejected, while a healthy routable CN remains reusable. No cache
capacity,
authentication, FIFO, timeout, routing-label, or circuit-breaker state
semantics
change. Regression tests cover nil, healthy, breaker-unhealthy,
recovered,
draining, and removed states.

## Test

`.agents/skills/mo-dev/scripts/mo-cgo-test -count=100 -timeout=5m -run
'^TestCanReuseCachedCNRequiresCurrentRouteEligibility$' ./pkg/proxy`

`.agents/skills/mo-dev/scripts/mo-cgo-test -race -count=20 -timeout=5m
-run '^TestCanReuseCachedCNRequiresCurrentRouteEligibility$'
./pkg/proxy`

`.agents/skills/mo-dev/scripts/mo-cgo-test -count=1 -timeout=10m
./pkg/proxy`

`.agents/skills/mo-dev/scripts/mo-cgo-test -race -count=1 -timeout=15m
./pkg/proxy`

`go build -mod=readonly ./pkg/proxy`

`go vet -mod=readonly ./pkg/proxy`

`git diff --check`

---------

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: LeftHandCold

pkg/proxy/client_conn.go

@@ -999,6 +999,7 @@ func (c *clientConn) connectToBackendContext(
 				c.connID,
 				c.mysqlProto.GetSalt(),
 				c.mysqlProto.GetAuthResponse(),
+				c.clientInfo,
 			)
 			cancel()
 		} else {
@@ -1007,6 +1008,7 @@ func (c *clientConn) connectToBackendContext(
 				c.connID,
 				c.mysqlProto.GetSalt(),
 				c.mysqlProto.GetAuthResponse(),
+				c.clientInfo,
 			)
 		}
 		if sc != nil {

pkg/proxy/client_conn_test.go

@@ -231,7 +231,9 @@ func (m *mockConnCache) Push(key cacheKey, sc ServerConn) bool {
 	return true
 }
 
-func (m *mockConnCache) Pop(key cacheKey, connID uint32, salt, authResp []byte) ServerConn {
+func (m *mockConnCache) Pop(
+	key cacheKey, connID uint32, salt, authResp []byte, _ clientInfo,
+) ServerConn {
 	if m.popFn != nil {
 		return m.popFn(key, connID, salt, authResp)
 	}
@@ -2414,6 +2416,63 @@ func Test_connectToBackend_SkipCacheOnMigration(t *testing.T) {
 	require.Equal(t, 0, cache.popCount)
 }
 
+func Test_connectToBackend_PassesClientInfoToCache(t *testing.T) {
+	cache := &popCountConnCache{}
+	client := clientInfo{labelInfo: labelInfo{
+		Tenant: "tenant1",
+		Labels: map[string]string{"region": "cn"},
+	}}
+	cConn := &clientConn{
+		ctx:        context.Background(),
+		router:     &routeErrRouter{},
+		mysqlProto: &frontend.MysqlProtocolImpl{},
+		connCache:  cache,
+		clientInfo: client,
+		log:        runtime.DefaultRuntime().Logger(),
+	}
+
+	sConn, err := cConn.connectToBackend("")
+	require.Error(t, err)
+	require.Nil(t, sConn)
+	require.Equal(t, 1, cache.popCount)
+	require.Equal(t, client, cache.lastClient)
+}
+
+type contextPopCountConnCache struct {
+	popCountConnCache
+	popContextCount int
+}
+
+func (c *contextPopCountConnCache) PopContext(
+	_ context.Context, _ cacheKey, _ uint32, _ []byte, _ []byte, client clientInfo,
+) ServerConn {
+	c.popContextCount++
+	c.lastClient = client
+	return nil
+}
+
+func Test_connectToBackend_PassesClientInfoToContextCache(t *testing.T) {
+	cache := &contextPopCountConnCache{}
+	client := clientInfo{labelInfo: labelInfo{
+		Tenant: "tenant1",
+		Labels: map[string]string{"region": "cn"},
+	}}
+	cConn := &clientConn{
+		ctx:        context.Background(),
+		router:     &routeErrRouter{},
+		mysqlProto: &frontend.MysqlProtocolImpl{},
+		connCache:  cache,
+		clientInfo: client,
+		log:        runtime.DefaultRuntime().Logger(),
+	}
+
+	sConn, err := cConn.connectToBackend("")
+	require.Error(t, err)
+	require.Nil(t, sConn)
+	require.Equal(t, 1, cache.popContextCount)
+	require.Equal(t, client, cache.lastClient)
+}
+
 func Test_connectToBackend_SkipCacheWhenPluginRouterEnabled(t *testing.T) {
 	cache := &popCountConnCache{}
 	cc, cleanup := createNewClientConn(t)

pkg/proxy/conn_cache.go

@@ -188,7 +188,7 @@ type ConnCache interface {
 	// Returns if the connection is pushed into the cache.
 	Push(cacheKey, ServerConn) bool
 	// Pop pops a server connection from the cache.
-	Pop(cacheKey, uint32, []byte, []byte) ServerConn
+	Pop(cacheKey, uint32, []byte, []byte, clientInfo) ServerConn
 	// Count returns the total number of cached connections. It is
 	// mainly for testing.
 	Count() int
@@ -197,7 +197,7 @@ type ConnCache interface {
 }
 
 type contextConnCache interface {
-	PopContext(context.Context, cacheKey, uint32, []byte, []byte) ServerConn
+	PopContext(context.Context, cacheKey, uint32, []byte, []byte, clientInfo) ServerConn
 }
 
 // the main cache struct.
@@ -234,7 +234,7 @@ type connCache struct {
 	// canReuseCN decides whether a cached connection to the CN may be reused
 	// for a fresh client login. If it returns false, the cached connection is
 	// discarded before any SET CONNECTION ID or auth work is attempted.
-	canReuseCN func(*CNServer) bool
+	canReuseCN func(*CNServer, clientInfo) bool
 }
 
 // connCacheOption is the option for connCache.
@@ -282,7 +282,7 @@ func withQueryClient(qc client.QueryClient) connCacheOption {
 	}
 }
 
-func withCanReuseCN(f func(*CNServer) bool) connCacheOption {
+func withCanReuseCN(f func(*CNServer, clientInfo) bool) connCacheOption {
 	return func(c *connCache) {
 		c.canReuseCN = f
 	}
@@ -418,10 +418,12 @@ func (c *connCache) closeCachedConnection(sc *serverConnAuth) {
 }
 
 // Pop implements the ConnCache interface.
-func (c *connCache) Pop(key cacheKey, connID uint32, salt []byte, authResp []byte) ServerConn {
+func (c *connCache) Pop(
+	key cacheKey, connID uint32, salt []byte, authResp []byte, client clientInfo,
+) ServerConn {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTransferTimeout)
 	defer cancel()
-	return c.PopContext(ctx, key, connID, salt, authResp)
+	return c.PopContext(ctx, key, connID, salt, authResp, client)
 }
 
 func (c *connCache) PopContext(
@@ -430,6 +432,7 @@ func (c *connCache) PopContext(
 	connID uint32,
 	salt []byte,
 	authResp []byte,
+	client clientInfo,
 ) ServerConn {
 	if ctx == nil {
 		ctx = context.Background()
@@ -463,16 +466,15 @@ func (c *connCache) PopContext(
 		}
 
 		// Before using a cached connection for a fresh login, ensure its CN is
-		// still eligible under the current health policy. This keeps connCache
-		// reuse aligned with the same breaker decision that Route() applies to
-		// newly-built sessions, and avoids executing SET CONNECTION ID against a
-		// CN that is currently cooling down.
-		if c.canReuseCN != nil && !c.canReuseCN(sc.GetCNServer()) {
+		// still eligible under the current login's route and health policy. This
+		// avoids executing SET CONNECTION ID against a CN that a fresh Route()
+		// would reject.
+		if c.canReuseCN != nil && !c.canReuseCN(sc.GetCNServer(), client) {
 			cnUUID := ""
 			if cn := sc.GetCNServer(); cn != nil {
 				cnUUID = cn.uuid
 			}
-			c.logger.Warn("skip cached connection on unhealthy cn",
+			c.logger.Warn("skip cached connection on ineligible cn",
 				zap.Uint32("conn ID", sc.ConnID()),
 				zap.String("cn", cnUUID),
 			)

pkg/proxy/conn_cache_test.go

@@ -301,17 +301,19 @@ func TestConnCache(t *testing.T) {
 			assert.True(t, cc.Push("k100", mockConn1))
 			assert.Equal(t, 1, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			sc := cc.Pop("k100", 1, nil, nil, clientInfo{})
 			assert.NotNil(t, sc)
 			assert.Equal(t, 0, cc.Count())
 		})
 	})
 
-	t.Run("pop - skip unhealthy cn before reuse", func(t *testing.T) {
+	t.Run("pop - skip route-ineligible cn before reuse", func(t *testing.T) {
 		runTestWithNewConnCacheWithAuthConstructor(t, nil, func(cc ConnCache) {
 			ccc := cc.(*connCache)
-			ccc.canReuseCN = func(cn *CNServer) bool {
-				return cn == nil || cn.uuid != "bad-cn"
+			var checkedClient clientInfo
+			ccc.canReuseCN = func(cn *CNServer, client clientInfo) bool {
+				checkedClient = client
+				return cn == nil || cn.uuid != "bad-cn" || client.username == "root"
 			}
 
 			c1, _ := net.Pipe()
@@ -320,9 +322,11 @@ func TestConnCache(t *testing.T) {
 			assert.True(t, cc.Push("k100", mockConn1))
 			assert.Equal(t, 1, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			login := clientInfo{username: "ordinary"}
+			sc := cc.Pop("k100", 1, nil, nil, login)
 			assert.Nil(t, sc)
 			assert.Equal(t, 0, cc.Count())
+			assert.Equal(t, login.username, checkedClient.username)
 		})
 	})
 
@@ -335,7 +339,7 @@ func TestConnCache(t *testing.T) {
 			assert.True(t, cc.Push("k100", mockConn1))
 			assert.Equal(t, 1, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			sc := cc.Pop("k100", 1, nil, nil, clientInfo{})
 			assert.Nil(t, sc)
 			assert.Equal(t, 0, cc.Count())
 		})
@@ -348,7 +352,7 @@ func TestConnCache(t *testing.T) {
 			assert.True(t, cc.Push("k100", mockConn1))
 			assert.Equal(t, 1, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			sc := cc.Pop("k100", 1, nil, nil, clientInfo{})
 			assert.Nil(t, sc)
 			assert.Equal(t, 1, cc.Count())
 		})
@@ -361,7 +365,7 @@ func TestConnCache(t *testing.T) {
 			assert.True(t, cc.Push("k100", mockConn1))
 			assert.Equal(t, 1, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			sc := cc.Pop("k100", 1, nil, nil, clientInfo{})
 			// cannot get conn as timeout.
 			assert.Nil(t, sc)
 			// count is 0 because the connection has been removed.
@@ -379,7 +383,7 @@ func TestConnCache(t *testing.T) {
 			assert.NoError(t, cc.Close())
 			assert.Equal(t, 0, cc.Count())
 
-			sc := cc.Pop("k100", 1, nil, nil)
+			sc := cc.Pop("k100", 1, nil, nil, clientInfo{})
 			assert.Nil(t, sc)
 
 			c2, _ := net.Pipe()
@@ -404,7 +408,7 @@ func TestConnCacheBlockedPopDoesNotBlockOtherTenantsOrClose(t *testing.T) {
 
 	popDone := make(chan ServerConn, 1)
 	go func() {
-		popDone <- cache.Pop("tenant-a", 1, nil, nil)
+		popDone <- cache.Pop("tenant-a", 1, nil, nil, clientInfo{})
 	}()
 	select {
 	case <-blocked.entered:
@@ -457,7 +461,7 @@ func TestConnCachePopContextCancelsBackendValidation(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	popDone := make(chan ServerConn, 1)
 	go func() {
-		popDone <- cache.(*connCache).PopContext(ctx, "tenant-a", 1, nil, nil)
+		popDone <- cache.(*connCache).PopContext(ctx, "tenant-a", 1, nil, nil, clientInfo{})
 	}()
 	select {
 	case <-blocked.entered:
@@ -540,7 +544,7 @@ func TestConnCachePopWaitsForOriginGenerationCleanup(t *testing.T) {
 	result := make(chan ServerConn, 1)
 	go func() {
 		result <- cache.(*connCache).PopContext(
-			context.Background(), "tenant-a", 1, nil, nil,
+			context.Background(), "tenant-a", 1, nil, nil, clientInfo{},
 		)
 	}()
 	select {
@@ -570,7 +574,7 @@ func TestConnCachePopWaitsForOriginGenerationCleanup(t *testing.T) {
 	canceledResult := make(chan ServerConn, 1)
 	go func() {
 		canceledResult <- cache.(*connCache).PopContext(
-			cancelCtx, "tenant-b", 2, nil, nil,
+			cancelCtx, "tenant-b", 2, nil, nil, clientInfo{},
 		)
 	}()
 	select {

pkg/proxy/plugin.go

@@ -123,9 +123,9 @@ func (r *pluginRouter) RouteForTransfer(
 	}
 }
 
-func (r *pluginRouter) CanReuseCachedCN(cn *CNServer) bool {
+func (r *pluginRouter) CanReuseCachedCN(cn *CNServer, client clientInfo) bool {
 	if rr, ok := r.Router.(cacheReuseChecker); ok {
-		return rr.CanReuseCachedCN(cn)
+		return rr.CanReuseCachedCN(cn, client)
 	}
 	return true
 }

pkg/proxy/plugin_test.go

@@ -33,6 +33,17 @@ import (
 
 var _ Router = (*pluginRouter)(nil)
 
+type cacheReuseTestRouter struct {
+	routeErrRouter
+	client clientInfo
+	result bool
+}
+
+func (r *cacheReuseTestRouter) CanReuseCachedCN(_ *CNServer, client clientInfo) bool {
+	r.client = client
+	return r.result
+}
+
 type mockPlugin struct {
 	mockRecommendCNFn func(ctx context.Context, clientInfo clientInfo) (*plugin.Recommendation, error)
 }
@@ -133,6 +144,15 @@ func TestPluginRouter_PropagatesConnectionContext(t *testing.T) {
 	require.Equal(t, 1, legacy.routeSelectedCount)
 }
 
+func TestPluginRouterCanReuseCachedCNDelegatesClientInfo(t *testing.T) {
+	delegate := &cacheReuseTestRouter{result: false}
+	router := newPluginRouter("", delegate, nil)
+	client := clientInfo{labelInfo: labelInfo{Tenant: "tenant1"}}
+
+	require.False(t, router.CanReuseCachedCN(&CNServer{uuid: "cn1"}, client))
+	require.Equal(t, client, delegate.client)
+}
+
 func TestPluginRouter_SelectHonorsBreaker(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

pkg/proxy/router.go

@@ -134,7 +134,7 @@ type transferRouter interface {
 // backend connection is still eligible for a fresh client login. Cached reuse
 // must honor the same CN health policy as a new session route.
 type cacheReuseChecker interface {
-	CanReuseCachedCN(cn *CNServer) bool
+	CanReuseCachedCN(cn *CNServer, client clientInfo) bool
 }
 
 // RefreshableRouter is a router that can be refreshed to get latest route strategy
@@ -403,14 +403,51 @@ func (r *router) RouteForTransfer(
 	return s, nil
 }
 
-// CanReuseCachedCN implements cacheReuseChecker. Cached connections must only
-// be reused when the breaker is closed/absent; otherwise they would bypass the
-// new-session health gate.
-func (r *router) CanReuseCachedCN(cn *CNServer) bool {
+// CanReuseCachedCN implements cacheReuseChecker. Cached connections must pass
+// the same cluster-eligibility and health gates as a newly routed session.
+func (r *router) CanReuseCachedCN(cn *CNServer, client clientInfo) bool {
 	if cn == nil {
 		return true
 	}
-	return r.health.canReuseCachedCN(cn.uuid)
+	if !r.health.canReuseCachedCN(cn.uuid) || r.moCluster == nil {
+		return false
+	}
+
+	candidates := make([]metadata.CNService, 0)
+	r.moCluster.GetCNService(
+		clusterservice.NewSelector(),
+		func(service metadata.CNService) bool {
+			candidates = append(candidates, service)
+			return true
+		},
+	)
+	if len(candidates) == 0 {
+		return false
+	}
+
+	// Apply the exact fresh-session routing policy to the current metadata for
+	// all working CNs. The full snapshot is required because labeled and empty
+	// CNs form priority tiers: whether a cached fallback remains eligible can
+	// change when another CN is added or removed.
+	eligible := false
+	selector := client.labelInfo.genSelector(clusterservice.EQ_Globbing)
+	appendFn := func(service *metadata.CNService) {
+		if service.ServiceID == cn.uuid {
+			eligible = true
+		}
+	}
+	if client.isSuperTenant() {
+		if err := route.RouteForSuperTenantCandidates(
+			context.Background(), candidates, selector, client.username, nil, appendFn,
+		); err != nil {
+			return false
+		}
+	} else if err := route.RouteForCommonTenantCandidates(
+		context.Background(), candidates, selector, nil, appendFn,
+	); err != nil {
+		return false
+	}
+	return eligible
 }
 
 func (r *router) connect(