Refactor keyspace manager to support periodic updates and improve its interface

Makefile

@@ -124,7 +124,7 @@ UT_PACKAGES_DISPATCHER := ./pkg/sink/cloudstorage/... ./pkg/sink/mysql/... ./pkg
 UT_PACKAGES_MAINTAINER := ./maintainer/... ./pkg/scheduler/...
 UT_PACKAGES_COORDINATOR := ./coordinator/...
 UT_PACKAGES_LOGSERVICE := ./logservice/...
-UT_PACKAGES_OTHERS := ./pkg/eventservice/... ./pkg/version/... ./utils/dynstream/... ./pkg/common/event/... ./pkg/common/... ./api/middleware/...
+UT_PACKAGES_OTHERS := ./pkg/eventservice/... ./pkg/version/... ./utils/dynstream/... ./pkg/common/event/... ./pkg/common/... ./api/middleware/... ./pkg/keyspace/...
 
 include tools/Makefile
 
@@ -281,6 +281,7 @@ unit_test_in_verify_ci_next_gen: check_failpoint_ctl tools/bin/gotestsum tools/b
 	tools/bin/gocov convert "$(TEST_DIR)/cov.unit.out" | tools/bin/gocov-xml > cdc-coverage.xml
 	$(FAILPOINT_DISABLE)
 
+# Usage: PKG=./api/middleware/... make unit_test_pkg
 unit_test_pkg: check_failpoint_ctl tools/bin/gotestsum tools/bin/gocov tools/bin/gocov-xml
 	mkdir -p "$(TEST_DIR)"
 	$(FAILPOINT_ENABLE)
@@ -294,6 +295,7 @@ unit_test_pkg: check_failpoint_ctl tools/bin/gotestsum tools/bin/gocov tools/bin
 	tools/bin/gocov convert "$(TEST_DIR)/cov.unit.out" | tools/bin/gocov-xml > cdc-coverage.xml
 	$(FAILPOINT_DISABLE)
 
+# Usage: PKG=./api/middleware/... make unit_test_pkg_next_gen
 unit_test_pkg_next_gen: check_failpoint_ctl tools/bin/gotestsum tools/bin/gocov tools/bin/gocov-xml
 	mkdir -p "$(TEST_DIR)"
 	$(FAILPOINT_ENABLE)

api/middleware/authenticate_middleware.go

@@ -102,7 +102,7 @@ func verify(ctx *gin.Context, etcdCli etcd.Client) error {
 
 // fetchTiDBTopology parses the TiDB topology from etcd.
 func fetchTiDBTopology(ctx context.Context, etcdClient etcd.Client, ks string) ([]upstream.TidbInstance, error) {
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	meta, err := keyspaceManager.LoadKeyspace(ctx, ks)
 	if err != nil {
 		return nil, err

api/middleware/middleware.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
-	"github.com/pingcap/kvproto/pkg/keyspacepb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/api"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
@@ -238,8 +237,8 @@ func KeyspaceCheckerMiddleware() gin.HandlerFunc {
 			return
 		}
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
-		meta, err := keyspaceManager.LoadKeyspace(c.Request.Context(), ks)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
+		_, err := keyspaceManager.ForceLoadKeyspace(c.Request.Context(), ks)
 		if errors.IsKeyspaceNotExistError(err) {
 			c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)
 			c.Abort()
@@ -250,12 +249,6 @@ func KeyspaceCheckerMiddleware() gin.HandlerFunc {
 			return
 		}
 
-		if meta.State != keyspacepb.KeyspaceState_ENABLED {
-			c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)
-			c.Abort()
-			return
-		}
-
 		c.Next()
 	}
 }

api/middleware/middleware_nextgen_test.go

@@ -37,7 +37,7 @@ func TestKeyspaceCheckerMiddleware(t *testing.T) {
 	tests := []struct {
 		name                 string
 		keyspace             string
-		init                 func(t *testing.T, mock *keyspace.MockKeyspaceManager)
+		init                 func(t *testing.T, mock *keyspace.MockManager)
 		expectedStatus       int
 		expectedAbort        bool
 		expectedBodyContains string
@@ -51,8 +51,8 @@ func TestKeyspaceCheckerMiddleware(t *testing.T) {
 		{
 			name:     "keyspace not exist",
 			keyspace: "not-exist",
-			init: func(t *testing.T, mock *keyspace.MockKeyspaceManager) {
-				mock.EXPECT().LoadKeyspace(gomock.Any(), "not-exist").Return(nil, errors.New(pdpb.ErrorType_ENTRY_NOT_FOUND.String()))
+			init: func(t *testing.T, mock *keyspace.MockManager) {
+				mock.EXPECT().ForceLoadKeyspace(gomock.Any(), "not-exist").Return(nil, errors.New(pdpb.ErrorType_ENTRY_NOT_FOUND.String()))
 			},
 			expectedStatus:       http.StatusBadRequest,
 			expectedAbort:        true,
@@ -61,8 +61,8 @@ func TestKeyspaceCheckerMiddleware(t *testing.T) {
 		{
 			name:     "internal server error",
 			keyspace: "internal-error",
-			init: func(t *testing.T, mock *keyspace.MockKeyspaceManager) {
-				mock.EXPECT().LoadKeyspace(gomock.Any(), "internal-error").Return(nil, errors.New("internal error"))
+			init: func(t *testing.T, mock *keyspace.MockManager) {
+				mock.EXPECT().ForceLoadKeyspace(gomock.Any(), "internal-error").Return(nil, errors.New("internal error"))
 			},
 			expectedStatus:       http.StatusInternalServerError,
 			expectedAbort:        true,
@@ -71,8 +71,8 @@ func TestKeyspaceCheckerMiddleware(t *testing.T) {
 		{
 			name:     "success",
 			keyspace: "success",
-			init: func(t *testing.T, mock *keyspace.MockKeyspaceManager) {
-				mock.EXPECT().LoadKeyspace(gomock.Any(), "success").Return(&keyspacepb.KeyspaceMeta{
+			init: func(t *testing.T, mock *keyspace.MockManager) {
+				mock.EXPECT().ForceLoadKeyspace(gomock.Any(), "success").Return(&keyspacepb.KeyspaceMeta{
 					State: keyspacepb.KeyspaceState_ENABLED,
 				}, nil)
 			},
@@ -87,7 +87,7 @@ func TestKeyspaceCheckerMiddleware(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 
-			mock := keyspace.NewMockKeyspaceManager(ctrl)
+			mock := keyspace.NewMockManager(ctrl)
 
 			if tt.init != nil {
 				tt.init(t, mock)

api/v2/changefeed.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
+	"github.com/pingcap/kvproto/pkg/keyspacepb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/api/middleware"
 	"github.com/pingcap/ticdc/downstreamadapter/sink"
@@ -103,13 +104,19 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 		return
 	}
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		_ = c.Error(errors.WrapError(errors.ErrKeyspaceNotFound, err))
 		return
 	}
 
+	if keyspaceMeta.State != keyspacepb.KeyspaceState_ENABLED {
+		c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)
+		c.Abort()
+		return
+	}
+
 	co, err := h.server.GetCoordinator()
 	if err != nil {
 		_ = c.Error(err)
@@ -381,7 +388,7 @@ func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
 	}
 	protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(replicaCfg.Sink.Protocol))
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceName := GetKeyspaceValueWithDefault(c)
 	kvStorage, err := keyspaceManager.GetStorage(keyspaceName)
 	if err != nil {
@@ -663,13 +670,19 @@ func (h *OpenAPIV2) ResumeChangefeed(c *gin.Context) {
 		newCheckpointTs = cfg.OverwriteCheckpointTs
 	}
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		_ = c.Error(errors.WrapError(errors.ErrKeyspaceNotFound, err))
 		return
 	}
 
+	if keyspaceMeta.State != keyspacepb.KeyspaceState_ENABLED {
+		c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)
+		c.Abort()
+		return
+	}
+
 	resumeGcServiceID := h.server.GetEtcdClient().GetEnsureGCServiceID(gc.EnsureGCServiceResuming)
 	if err := verifyResumeChangefeedConfig(
 		ctx,
@@ -750,6 +763,18 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 		return
 	}
 
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
+	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
+	if err != nil {
+		_ = c.Error(errors.WrapError(errors.ErrKeyspaceNotFound, err))
+		return
+	}
+	if keyspaceMeta.State != keyspacepb.KeyspaceState_ENABLED {
+		c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)
+		c.Abort()
+		return
+	}
+
 	oldCfInfo, status, err := co.GetChangefeed(c, changefeedDisplayName)
 	if err != nil {
 		_ = c.Error(err)
@@ -820,8 +845,6 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 		}
 		protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(oldCfInfo.Config.Sink.Protocol))
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
-
 		kvStorage, err := keyspaceManager.GetStorage(keyspaceName)
 		if err != nil {
 			_ = c.Error(err)

api/v2/unsafe.go

@@ -54,7 +54,7 @@ func (h *OpenAPIV2) ResolveLock(c *gin.Context) {
 
 	keyspaceName := GetKeyspaceValueWithDefault(c)
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	// The ctx's lifecycle is the same as the HTTP request.
 	// The schema store may use the context to fetch database information asynchronously.
 	// Therefore, we cannot use the context of the HTTP request.
@@ -93,7 +93,7 @@ func (h *OpenAPIV2) DeleteServiceGcSafePoint(c *gin.Context) {
 	defer pdClient.Close()
 
 	keyspaceName := GetKeyspaceValueWithDefault(c)
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(c.Request.Context(), keyspaceName)
 	if err != nil {
 		_ = c.Error(cerror.WrapError(cerror.ErrKeyspaceNotFound, err))

coordinator/coordinator.go

@@ -287,7 +287,7 @@ func (c *coordinator) handleStateChange(
 		log.Info("changefeed is resumed or created successfully, try to delete its safeguard gc safepoint",
 			zap.String("changefeed", event.changefeedID.String()))
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 		keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, event.changefeedID.Keyspace())
 		if err != nil {
 			log.Warn("failed to load keyspace", zap.String("keyspace", event.changefeedID.Keyspace()), zap.Error(err))
@@ -473,7 +473,7 @@ func (c *coordinator) updateAllKeyspaceGcBarriers(ctx context.Context) error {
 
 func (c *coordinator) updateKeyspaceGcBarrier(ctx context.Context, barrierMap map[string]uint64, keyspaceName string) error {
 	// Obtain keyspace metadata from PD
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		return cerror.WrapError(cerror.ErrLoadKeyspaceFailed, err)

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -158,7 +158,7 @@ func NewDispatcherManager(
 	ctx, cancel := context.WithCancel(context.Background())
 	pdClock := appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock)
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, changefeedID.Keyspace())
 	if err != nil {
 		cancel()

logservice/schemastore/schema_store.go

@@ -250,7 +250,7 @@ func (s *schemaStore) getKeyspaceSchemaStore(keyspaceID uint32) (*keyspaceSchema
 
 	// If the schemastore does not contain the keyspace, it means it is not a maintainer node.
 	// It should register the keyspace when it try to get keyspace schema_store.
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.GetKeyspaceByID(ctx, keyspaceID)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -443,7 +443,7 @@ func (s *schemaStore) RegisterKeyspace(
 	s.keyspaceLocker.Lock()
 	defer s.keyspaceLocker.Unlock()
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		return errors.Trace(err)

logservice/txnutil/lock_resolver.go

@@ -69,7 +69,7 @@ func (r *resolver) Resolve(ctx context.Context, keyspaceID uint32, regionID uint
 		Limit:      scanLockLimit,
 	})
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.GetKeyspaceByID(ctx, keyspaceID)
 	if err != nil {
 		return err

maintainer/maintainer_manager.go

@@ -231,7 +231,7 @@ func (m *Manager) onAddMaintainerRequest(req *heartbeatpb.AddMaintainerRequest)
 	}
 
 	ctx := context.Background()
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, cfID.Keyspace())
 	if err != nil {
 		// BUG tenfyzhong 2025-09-11 17:29:08 how to process err
@@ -261,7 +261,7 @@ func (m *Manager) onRemoveMaintainerRequest(msg *messaging.TargetMessage) *heart
 		}
 
 		ctx := context.Background()
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 		keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, cfID.Keyspace())
 		if err != nil {
 			// BUG tenfyzhong 2025-09-11 17:29:08 how to process err

maintainer/maintainer_manager_test.go

@@ -77,7 +77,7 @@ func TestMaintainerSchedulesNodeChanges(t *testing.T) {
 			Name: "ks1",
 		}
 	}
-	keyspaceManager := keyspace.NewMockKeyspaceManager(ctrl)
+	keyspaceManager := keyspace.NewMockManager(ctrl)
 	keyspaceManager.EXPECT().LoadKeyspace(gomock.Any(), gomock.Any()).Return(meta, nil)
 	appcontext.SetService(appcontext.KeyspaceManager, keyspaceManager)
 
@@ -304,7 +304,7 @@ func TestMaintainerBootstrapWithTablesReported(t *testing.T) {
 			Name: "ks1",
 		}
 	}
-	keyspaceManager := keyspace.NewMockKeyspaceManager(ctrl)
+	keyspaceManager := keyspace.NewMockManager(ctrl)
 	keyspaceManager.EXPECT().LoadKeyspace(gomock.Any(), gomock.Any()).Return(meta, nil)
 	appcontext.SetService(appcontext.KeyspaceManager, keyspaceManager)
 
@@ -440,7 +440,7 @@ func TestStopNotExistsMaintainer(t *testing.T) {
 			Name: "ks1",
 		}
 	}
-	keyspaceManager := keyspace.NewMockKeyspaceManager(ctrl)
+	keyspaceManager := keyspace.NewMockManager(ctrl)
 	keyspaceManager.EXPECT().LoadKeyspace(gomock.Any(), gomock.Any()).Return(meta, nil).AnyTimes()
 
 	appcontext.SetService(appcontext.KeyspaceManager, keyspaceManager)