fix: fail closed when draining CN pipelines

api/core/v1alpha1/semver.go

@@ -35,7 +35,7 @@ var (
 	// the new major branch before its gate is extended, to avoid unintentionally flipping
 	// behavior (e.g. lock migration handshake, pipeline/sharding stats collection).
 	featureVersions = map[MOFeature][]semver.Version{
-		MOFeaturePipelineInfo:      {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0")},
+		MOFeaturePipelineInfo:      {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0"), semver.MustParse("3.0.0"), semver.MustParse("4.0.0")},
 		MOFeatureSessionSource:     {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0")},
 		MOFeatureLockMigration:     {semver.MustParse("1.2.0"), semver.MustParse("2.0.0")},
 		MOFeatureShardingMigration: {semver.MustParse("2.0.0")},

api/core/v1alpha1/semver_test.go

@@ -41,6 +41,13 @@ func TestHasMOFeature(t *testing.T) {
 	g.Expect(HasMOFeature(mustParse("1.1.3"), MOFeaturePipelineInfo)).To(BeTrue())
 	g.Expect(HasMOFeature(mustParse("1.2.0"), MOFeaturePipelineInfo)).To(BeTrue())
 	g.Expect(HasMOFeature(mustParse("2.0.0"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("3.0.0"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("3.9.9"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("v3.0.10-90ae8dc0c-2026-04-22"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("4.0.0"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("v4.0.0-rc1"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("4.2.1"), MOFeaturePipelineInfo)).To(BeTrue())
+	g.Expect(HasMOFeature(mustParse("v4.2.1-d2393868a-2026-08-28"), MOFeaturePipelineInfo)).To(BeTrue())
 	g.Expect(HasMOFeature(mustParse("1.1.1"), MOFeaturePipelineInfo)).To(BeFalse())
 	g.Expect(HasMOFeature(mustParse("1.1.1"), MOFeaturePipelineInfo)).To(BeFalse())
 	g.Expect(HasMOFeature(mustParse("1.1.2"), MOFeaturePipelineInfo)).To(BeTrue())
@@ -79,14 +86,13 @@ func TestHasMOFeature_DiscoveryFixed(t *testing.T) {
 // TestHasMOFeature_OtherFeaturesNotExtendedTo3x guards against accidentally widening the
 // version gate for features that have NOT been explicitly verified against MO 3.x. Extending
 // featureVersions in bulk (i.e. blindly adding "3.0.0" to every feature) would silently flip
-// unrelated behavior (lock migration handshake, pipeline/sharding stats collection, session
-// source accounting) on MO 3.x without dedicated verification. Only MOFeatureDiscoveryFixed
-// has been confirmed compatible with 3.x so far (see #597); this test should be updated
-// deliberately, one feature at a time, as each is verified.
+// unrelated behavior (lock migration handshake, sharding stats collection, session source
+// accounting) on MO 3.x without dedicated verification. DiscoveryFixed and PipelineInfo have
+// been verified independently; this test should be updated deliberately, one feature at a time,
+// as each remaining feature is verified.
 func TestHasMOFeature_OtherFeaturesNotExtendedTo3x(t *testing.T) {
 	g := NewGomegaWithT(t)
 	unverifiedOn3x := []MOFeature{
-		MOFeaturePipelineInfo,
 		MOFeatureSessionSource,
 		MOFeatureLockMigration,
 		MOFeatureShardingMigration,

pkg/controllers/cnstore/controller.go

@@ -34,6 +34,7 @@ import (
 	"github.com/matrixorigin/matrixone-operator/pkg/querycli"
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	querypb "github.com/matrixorigin/matrixone/pkg/pb/query"
 	"github.com/openkruise/kruise-api/apps/pub"
 	kruisev1alpha1 "github.com/openkruise/kruise-api/apps/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
@@ -69,7 +70,13 @@ const resyncInterval = 30 * time.Second
 
 type Controller struct {
 	clientMgr *mocli.MORPCClientManager
-	queryCli  *querycli.Client
+	queryCli  queryClient
+}
+
+type queryClient interface {
+	ShowProcessList(context.Context, string) (*querypb.ShowProcessListResponse, error)
+	GetPipelineInfo(context.Context, string) (*querypb.GetPipelineInfoResponse, error)
+	GetReplicaCount(context.Context, string) (querypb.GetReplicaCountResponse, error)
 }
 
 type withCNSet struct {
@@ -492,10 +499,9 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error {
 	if err == nil {
 		sc = previous
 	}
-	if sc.StartedTime == nil || !sc.StartedTime.Equal(*startedTime) {
-		// clean previously recorded score and update startTime if CN is restarted
-		sc.Restarted(startedTime)
-	}
+	// Invalidate the previous round before performing any query. A failed or partial
+	// refresh must never leave an old zero looking like current evidence.
+	sc.BeginObservation(startedTime)
 
 	uid := v1alpha1.GetCNPodUUID(pod)
 	moVersion := common.GetSemanticVersion(&pod.ObjectMeta)
@@ -509,6 +515,9 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error {
 		return nil
 	}); err != nil {
 		ctx.Log.Info("error refresh stats, cn not found in store-cache", "error", err.Error())
+		// BeginObservation has already invalidated this round. Persist that state so
+		// IsSafeToReclaim remains the single reclaim-safety boundary, while preserving
+		// the existing state-sync behavior when the CN is absent from the cache.
 		return c.patchStoreStats(ctx, sc)
 	}
 
@@ -517,39 +526,52 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error {
 		Logger:  ctx.Log,
 		Enabled: diagnosDraining,
 	}
+	c.collectQueryStats(sc, queryAddress, moVersion, diagosis)
+
+	return c.patchStoreStats(ctx, sc)
+}
+
+// collectQueryStats updates one observation round. Each observation is marked
+// valid only after its query succeeds. A feature that is unavailable for the
+// current MO version is explicitly treated as not required with a zero count.
+func (c *Controller) collectQueryStats(sc *common.StoreScore, queryAddress string, moVersion semver.Version, diagosis *connectionDiagnosis) {
 	count, err := c.getSessionCount(queryAddress, moVersion, diagosis)
 	if err != nil {
-		ctx.Log.Info("error get session count", "error", err.Error())
+		diagosis.Logger.Info("error get session count", "error", err.Error())
 	} else {
 		// update session count
 		sc.SessionCount = count
+		sc.SessionObserved = true
 	}
 	var pipelineCount int
 	if v1alpha1.HasMOFeature(moVersion, v1alpha1.MOFeaturePipelineInfo) {
 		pipelineCount, err = c.getPipelineCount(queryAddress, diagosis)
 		if err != nil {
-			ctx.Log.Info("error get pipeline count", "error", err.Error())
+			diagosis.Logger.Info("error get pipeline count", "error", err.Error())
 		} else {
 			// update pipeline count
 			sc.PipelineCount = pipelineCount
+			sc.PipelineObserved = true
 		}
 	} else {
-		// clear pipeline count if feature is disabled
+		// Pipeline observation is not required for versions without the API.
 		sc.PipelineCount = 0
+		sc.PipelineObserved = true
 	}
 	var replicaCount int
 	if v1alpha1.HasMOFeature(moVersion, v1alpha1.MOFeatureShardingMigration) {
 		replicaCount, err = c.getReplicaCount(queryAddress, diagosis)
 		if err != nil {
-			ctx.Log.Info("error get replica count", "error", err.Error())
+			diagosis.Logger.Info("error get replica count", "error", err.Error())
 		} else {
 			sc.ReplicaCount = replicaCount
+			sc.ReplicaObserved = true
 		}
 	} else {
+		// Replica observation is not required for versions without sharding migration.
 		sc.ReplicaCount = 0
+		sc.ReplicaObserved = true
 	}
-
-	return c.patchStoreStats(ctx, sc)
 }
 
 func (c *Controller) patchStoreStats(ctx *recon.Context[*corev1.Pod], sc *common.StoreScore) error {

pkg/controllers/cnstore/stats_test.go

@@ -0,0 +1,125 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cnstore
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/blang/semver/v4"
+	"github.com/go-logr/logr"
+	"github.com/matrixorigin/matrixone-operator/pkg/controllers/common"
+	querypb "github.com/matrixorigin/matrixone/pkg/pb/query"
+)
+
+type fakeQueryClient struct {
+	pipelineCount int64
+	replicaCount  int64
+	sessionErr    error
+	pipelineErr   error
+	replicaErr    error
+
+	pipelineCalls int
+	replicaCalls  int
+}
+
+func (f *fakeQueryClient) ShowProcessList(context.Context, string) (*querypb.ShowProcessListResponse, error) {
+	if f.sessionErr != nil {
+		return nil, f.sessionErr
+	}
+	return &querypb.ShowProcessListResponse{}, nil
+}
+
+func (f *fakeQueryClient) GetPipelineInfo(context.Context, string) (*querypb.GetPipelineInfoResponse, error) {
+	f.pipelineCalls++
+	if f.pipelineErr != nil {
+		return nil, f.pipelineErr
+	}
+	return &querypb.GetPipelineInfoResponse{Count: f.pipelineCount}, nil
+}
+
+func (f *fakeQueryClient) GetReplicaCount(context.Context, string) (querypb.GetReplicaCountResponse, error) {
+	f.replicaCalls++
+	if f.replicaErr != nil {
+		return querypb.GetReplicaCountResponse{}, f.replicaErr
+	}
+	return querypb.GetReplicaCountResponse{Count: f.replicaCount}, nil
+}
+
+func TestCollectQueryStatsFailClosed(t *testing.T) {
+	tests := []struct {
+		name          string
+		version       string
+		query         *fakeQueryClient
+		safe          bool
+		pipelineCalls int
+		replicaCalls  int
+	}{
+		{
+			name:          "MO 4 complete zero",
+			version:       "4.2.0",
+			query:         &fakeQueryClient{},
+			safe:          true,
+			pipelineCalls: 1,
+		},
+		{
+			name:          "MO 4 active AP pipeline",
+			version:       "4.2.0",
+			query:         &fakeQueryClient{pipelineCount: 1},
+			pipelineCalls: 1,
+		},
+		{
+			name:          "MO 4 pipeline query failure",
+			version:       "4.2.0",
+			query:         &fakeQueryClient{pipelineErr: errors.New("pipeline unavailable")},
+			pipelineCalls: 1,
+		},
+		{
+			name:          "session query failure",
+			version:       "4.2.0",
+			query:         &fakeQueryClient{sessionErr: errors.New("session unavailable")},
+			pipelineCalls: 1,
+		},
+		{
+			name:          "required replica query failure",
+			version:       "2.0.0",
+			query:         &fakeQueryClient{replicaErr: errors.New("replica unavailable")},
+			pipelineCalls: 1,
+			replicaCalls:  1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			started := time.Now()
+			score := &common.StoreScore{StartedTime: &started}
+			score.BeginObservation(&started)
+			controller := &Controller{queryCli: tt.query}
+			controller.collectQueryStats(score, "query-address", semver.MustParse(tt.version), &connectionDiagnosis{Logger: logr.Discard()})
+
+			if got := score.IsSafeToReclaim(); got != tt.safe {
+				t.Fatalf("IsSafeToReclaim() = %v, want %v; score=%+v", got, tt.safe, score)
+			}
+			if tt.query.pipelineCalls != tt.pipelineCalls {
+				t.Fatalf("pipeline calls = %d, want %d", tt.query.pipelineCalls, tt.pipelineCalls)
+			}
+			if tt.query.replicaCalls != tt.replicaCalls {
+				t.Fatalf("replica calls = %d, want %d", tt.query.replicaCalls, tt.replicaCalls)
+			}
+		})
+	}
+}

pkg/controllers/common/cnstore.go

@@ -151,6 +151,12 @@ type StoreScore struct {
 	SessionCount  int `json:"sessionCount"`
 	PipelineCount int `json:"pipelineCount"`
 	ReplicaCount  int `json:"replicaCount"`
+	// The observed fields distinguish a confirmed count from a zero value left by a
+	// skipped, failed, or incomplete query. Old annotations do not contain these
+	// fields and are deliberately unsafe until a fresh observation succeeds.
+	SessionObserved  bool `json:"sessionObserved"`
+	PipelineObserved bool `json:"pipelineObserved"`
+	ReplicaObserved  bool `json:"replicaObserved"`
 
 	StartedTime *time.Time `json:"startedTime,omitempty"`
 }
@@ -160,14 +166,23 @@ func (s *StoreScore) GenDeletionCost() int {
 }
 
 func (s *StoreScore) IsSafeToReclaim() bool {
-	return s.SessionCount == 0 && s.PipelineCount == 0 && s.ReplicaCount == 0
-}
-
-func (s *StoreScore) Restarted(startedTime *time.Time) {
-	s.SessionCount = 0
-	s.PipelineCount = 0
-	s.ReplicaCount = 0
-	s.StartedTime = startedTime
+	return s.SessionObserved && s.PipelineObserved && s.ReplicaObserved &&
+		s.SessionCount == 0 && s.PipelineCount == 0 && s.ReplicaCount == 0
+}
+
+// BeginObservation invalidates the previous round before any query starts. Counts
+// are kept for diagnostics and deletion cost unless the CN process restarted, but
+// they cannot authorize reclaim until every required observation succeeds again.
+func (s *StoreScore) BeginObservation(startedTime *time.Time) {
+	s.SessionObserved = false
+	s.PipelineObserved = false
+	s.ReplicaObserved = false
+	if s.StartedTime == nil || !s.StartedTime.Equal(*startedTime) {
+		s.SessionCount = 0
+		s.PipelineCount = 0
+		s.ReplicaCount = 0
+		s.StartedTime = startedTime
+	}
 }
 
 // GetStoreScore get the store connection count from Pod anno