fix(frontend): sync global sysvar commits across CNs

pkg/clusterservice/cluster.go

@@ -30,6 +30,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 )
 
 // GetMOCluster get mo cluster from process level runtime
@@ -187,6 +188,16 @@ func WithDisableRefresh() Option {
 	}
 }
 
+// WithGlobalSysVarRoutingFilter makes the cluster snapshot suitable for SQL
+// routing by excluding CNs that have not applied the durable global-system-
+// variable watermark. It must not be enabled for the process-wide discovery
+// view used by lockservice and other internal components.
+func WithGlobalSysVarRoutingFilter() Option {
+	return func(c *cluster) {
+		c.options.globalSysVarRoutingFilter = true
+	}
+}
+
 type cluster struct {
 	logger          *log.MOLogger
 	stopper         *stopper.Stopper
@@ -207,11 +218,13 @@ type cluster struct {
 	// Correctness: readyOnce.Do guarantees that ready.Store(true) happens before
 	// close(readyC), so if readyC is closed (i.e., <-readyC returns), ready is
 	// guaranteed to be true. If ready.Load() returns false, we fall back to channel wait.
-	ready       atomic.Bool
-	services    atomic.Pointer[services]
-	regexpCache *regexpCache
-	options     struct {
-		disableRefresh bool
+	ready                atomic.Bool
+	services             atomic.Pointer[services]
+	globalSysVarCommitTS atomic.Pointer[timestamp.Timestamp]
+	regexpCache          *regexpCache
+	options              struct {
+		disableRefresh            bool
+		globalSysVarRoutingFilter bool
 	}
 }
 
@@ -526,6 +539,17 @@ func (c *cluster) refreshWithContext(ctx context.Context) error {
 
 	new := &services{}
 	for _, cn := range details.CNStores {
+		if c.options.globalSysVarRoutingFilter &&
+			!details.GlobalSysVarCommitTS.IsEmpty() &&
+			cn.GlobalSysVarCommitTS.Less(details.GlobalSysVarCommitTS) {
+			if c.logger.Enabled(zap.DebugLevel) {
+				c.logger.Debug("cn service fenced by global sysvar watermark",
+					zap.String("cn", cn.UUID),
+					zap.String("required", details.GlobalSysVarCommitTS.DebugString()),
+					zap.String("applied", cn.GlobalSysVarCommitTS.DebugString()))
+			}
+			continue
+		}
 		v := newCNService(cn)
 		new.addCN([]metadata.CNService{v})
 		if c.logger.Enabled(zap.DebugLevel) {
@@ -549,13 +573,41 @@ func (c *cluster) refreshWithContext(ctx context.Context) error {
 		new.tn = new.tn[:1]
 	}
 	c.services.Store(new)
+	c.publishGlobalSysVarCommitTS(details.GlobalSysVarCommitTS)
 	c.readyOnce.Do(func() {
 		c.ready.Store(true)
 		close(c.readyC)
 	})
 	return nil
 }
 
+func (c *cluster) publishGlobalSysVarCommitTS(ts timestamp.Timestamp) {
+	if ts.IsEmpty() {
+		return
+	}
+	for {
+		current := c.globalSysVarCommitTS.Load()
+		if current != nil && current.GreaterEq(ts) {
+			return
+		}
+		next := ts
+		if c.globalSysVarCommitTS.CompareAndSwap(current, &next) {
+			return
+		}
+	}
+}
+
+// GlobalSysVarCommitTS returns the durable routing watermark represented by
+// service's currently published CN snapshot.
+func GlobalSysVarCommitTS(service MOCluster) timestamp.Timestamp {
+	if c, ok := service.(*cluster); ok {
+		if ts := c.globalSysVarCommitTS.Load(); ts != nil {
+			return *ts
+		}
+	}
+	return timestamp.Timestamp{}
+}
+
 func (c *cluster) acquireRefresh(ctx context.Context) error {
 	c.refreshGateOnce.Do(func() {
 		c.refreshGateC = make(chan struct{}, 1)

pkg/clusterservice/cluster_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 )
 
 func TestClusterReady(t *testing.T) {
@@ -308,13 +309,21 @@ func TestCluster_GetTNService(t *testing.T) {
 func runClusterTest(
 	refreshInterval time.Duration,
 	fn func(*testHAKeeperClient, *cluster),
+) {
+	runClusterTestWithOptions(refreshInterval, fn)
+}
+
+func runClusterTestWithOptions(
+	refreshInterval time.Duration,
+	fn func(*testHAKeeperClient, *cluster),
+	opts ...Option,
 ) {
 	sid := ""
 	runtime.RunTest(
 		sid,
 		func(rt runtime.Runtime) {
 			hc := &testHAKeeperClient{}
-			c := NewMOCluster(sid, hc, refreshInterval)
+			c := NewMOCluster(sid, hc, refreshInterval, opts...)
 			defer c.Close()
 			fn(hc, c.(*cluster))
 		},
@@ -349,6 +358,63 @@ func (c *testHAKeeperClient) addTN(tick uint64, serviceIDs ...string) {
 	}
 }
 
+func TestClusterGlobalSysVarWatermarkFilteringIsRoutingOnly(t *testing.T) {
+	runClusterTestWithOptions(
+		time.Hour,
+		func(hc *testHAKeeperClient, c *cluster) {
+			commitTS := timestamp.Timestamp{PhysicalTime: 100, LogicalTime: 1}
+			hc.Lock()
+			hc.value = logpb.ClusterDetails{
+				GlobalSysVarCommitTS: commitTS,
+				CNStores: []logpb.CNStore{
+					{UUID: "cn-a", WorkState: metadata.WorkState_Working, GlobalSysVarCommitTS: commitTS},
+					{UUID: "cn-b", WorkState: metadata.WorkState_Working},
+				},
+			}
+			hc.Unlock()
+
+			c.ForceRefresh(true)
+			require.Equal(t, []string{"cn-a"}, routableCNIDs(c))
+
+			hc.Lock()
+			hc.value.CNStores[1].GlobalSysVarCommitTS = commitTS
+			hc.Unlock()
+			c.ForceRefresh(true)
+			require.ElementsMatch(t, []string{"cn-a", "cn-b"}, routableCNIDs(c))
+		},
+		WithGlobalSysVarRoutingFilter(),
+	)
+
+	runClusterTest(
+		time.Hour,
+		func(hc *testHAKeeperClient, c *cluster) {
+			commitTS := timestamp.Timestamp{PhysicalTime: 100, LogicalTime: 1}
+			hc.Lock()
+			hc.value = logpb.ClusterDetails{
+				GlobalSysVarCommitTS: commitTS,
+				CNStores: []logpb.CNStore{
+					{UUID: "cn-a", WorkState: metadata.WorkState_Working, GlobalSysVarCommitTS: commitTS},
+					{UUID: "cn-b", WorkState: metadata.WorkState_Working},
+				},
+			}
+			hc.Unlock()
+
+			c.ForceRefresh(true)
+			require.ElementsMatch(t, []string{"cn-a", "cn-b"}, routableCNIDs(c),
+				"general service discovery must not be filtered by SQL routing admission")
+		},
+	)
+}
+
+func routableCNIDs(c *cluster) []string {
+	var ids []string
+	c.GetCNService(NewSelector(), func(cn metadata.CNService) bool {
+		ids = append(ids, cn.ServiceID)
+		return true
+	})
+	return ids
+}
+
 func TestNewTNServicePreservesAutoIncrEpochFenceCapability(t *testing.T) {
 	service := newTNService(logpb.TNStore{
 		UUID:                        "tn-new",

pkg/cnservice/server.go

@@ -155,15 +155,17 @@ func NewService(
 			UUID: cfg.UUID,
 			Role: metadata.MustParseCNRole(cfg.Role),
 		},
-		cfg:         cfg,
-		logger:      logutil.GetGlobalLogger().Named("cn-service"),
-		metadataFS:  metadataFS,
-		etlFS:       etlFS,
-		fileService: fileService,
-		sessionMgr:  queryservice.NewSessionManager(),
-		addressMgr:  address.NewAddressManager(cfg.ServiceHost, cfg.PortBase),
-		gossipNode:  gossipNode,
-	}
+		cfg:                    cfg,
+		logger:                 logutil.GetGlobalLogger().Named("cn-service"),
+		metadataFS:             metadataFS,
+		etlFS:                  etlFS,
+		fileService:            fileService,
+		sessionMgr:             queryservice.NewSessionManager(),
+		addressMgr:             address.NewAddressManager(cfg.ServiceHost, cfg.PortBase),
+		gossipNode:             gossipNode,
+		globalSysVarGeneration: uuid.NewString(),
+	}
+	srv.initControlChannels()
 	srv.colexecServer = colexec.NewServer(cfg.UUID)
 
 	srv.requestHandler = func(ctx context.Context,
@@ -412,6 +414,15 @@ func (s *service) Start() (err error) {
 	if err = s.bootstrap(); err != nil {
 		return err
 	}
+	admissionCtx, admissionCancel := context.WithTimeout(
+		context.Background(), s.cfg.HAKeeper.DiscoveryTimeout.Duration)
+	err = s.waitGlobalSysVarAdmission(admissionCtx)
+	if err != nil {
+		err = moerr.AttachCause(admissionCtx, err)
+		admissionCancel()
+		return err
+	}
+	admissionCancel()
 
 	s.initSqlWriterFactory()