fix: fence DDL visibility across CN admission

pkg/clusterservice/cluster.go

@@ -655,6 +655,8 @@ func newCNService(cn logpb.CNStore) metadata.CNService {
 		ViewMetadataAdmissionGeneration: cn.ViewMetadataAdmissionGeneration,
 		ViewMetadataAdmissionReady:      cn.ViewMetadataAdmissionReady,
 		ViewMetadataObservedEpoch:       cn.ViewMetadataObservedEpoch,
+		DDLVisibilityBarrierReady:       cn.DDLVisibilityBarrierReady,
+		ViewMetadataIngressReady:        cn.ViewMetadataIngressReady,
 	}
 }
 

pkg/clusterservice/cluster_test.go

@@ -169,6 +169,8 @@ func TestClusterAdmissionSnapshotFiltersEveryPublicInventory(t *testing.T) {
 					UUID:                            "pending",
 					WorkState:                       metadata.WorkState_Working,
 					ViewMetadataAdmissionGeneration: 11,
+					DDLVisibilityBarrierReady:       true,
+					ViewMetadataIngressReady:        true,
 				},
 			}
 			hc.Unlock()
@@ -189,12 +191,20 @@ func TestClusterAdmissionSnapshotFiltersEveryPublicInventory(t *testing.T) {
 			require.Equal(t, []string{"ready"}, withoutWorkState)
 
 			var raw []string
+			pendingBarrierReady := false
+			pendingIngressReady := false
 			require.NoError(t, GetCNServiceRawWithContext(
 				context.Background(), c, NewSelector(), func(service metadata.CNService) bool {
 					raw = append(raw, service.ServiceID)
+					if service.ServiceID == "pending" {
+						pendingBarrierReady = service.DDLVisibilityBarrierReady
+						pendingIngressReady = service.ViewMetadataIngressReady
+					}
 					return true
 				}))
 			require.ElementsMatch(t, []string{"ready", "pending"}, raw)
+			require.True(t, pendingBarrierReady)
+			require.True(t, pendingIngressReady)
 
 			admission := c.GetViewMetadataAdmission()
 			require.True(t, admission.Enabled)

pkg/cnservice/server.go

@@ -155,15 +155,18 @@ 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,
+		ddlCommitGate: frontend.NewDDLCommitGate(),
+	}
+	runtime.ServiceRuntime(cfg.UUID).SetGlobalVariables(
+		frontend.DDLCommitGateRuntimeKey, srv.ddlCommitGate)
 	srv.colexecServer = colexec.NewServer(cfg.UUID)
 
 	srv.requestHandler = func(ctx context.Context,
@@ -463,6 +466,9 @@ func (s *service) Start() (err error) {
 	if err = s.startUnlessViewMetadataGenerationRevoked(s.queryService.Start); err != nil {
 		return err
 	}
+	if err = s.prepareDDLVisibilityBarrier(); err != nil {
+		return err
+	}
 	if err = s.startFrontendUnlessViewMetadataGenerationRevoked(); err != nil {
 		return err
 	}
@@ -477,11 +483,9 @@ func (s *service) Start() (err error) {
 	if err = s.checkViewMetadataGenerationRevoked(); err != nil {
 		return err
 	}
-	s.viewMetadataIngressReady.Store(true)
-	if err = s.checkViewMetadataGenerationRevoked(); err != nil {
+	if err = s.publishDDLVisibilityIngressAfterStart(); err != nil {
 		return err
 	}
-	s.notifyHeartbeat()
 
 	if err = s.checkViewMetadataGenerationRevoked(); err != nil {
 		return err
@@ -508,9 +512,20 @@ func (s *service) closeService() error {
 		defer logutil.LogClose(s.logger, "cnservice")()
 
 		s.closeViewMetadataAdmission()
+		// Prevent an already-admitted protocol command from republishing the
+		// barrier after shutdown begins. Other QueryService methods remain live
+		// until authoritative withdrawal, preserving the stale-target retry path.
+		s.ddlVisibilityBarrierClosing.Store(true)
+		if s.ddlCommitGate != nil {
+			s.ddlCommitGate.Close()
+		}
+		// Stop periodic heartbeats before publishing the final false readiness.
+		// QueryService remains available until the authoritative inventory has
+		// observed the withdrawal, so healthy CNs cannot target a closed barrier.
 		s.stopper.Stop()
 
 		s.closeErr = closeCNServiceSteps(
+			s.withdrawDDLVisibilityBarrier,
 			// Query commands can reach frontend, task, engine, lock, shard,
 			// auto-increment, and transaction state. Stop and drain this remote
 			// ingress before clearing any of those dependencies.