Merge pull request #18 from LiteyukiStudio/codex/fix-cluster-resource-drift

fix 🐛: 修复集群资源漂移与删除清理

Author: snowykami

internal/api/application_bridge.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"errors"
 
 	"github.com/LiteyukiStudio/devops/internal/api/applicationapi"
 	"github.com/LiteyukiStudio/devops/internal/api/buildapi"
@@ -11,6 +12,7 @@ import (
 	kubeprovider "github.com/LiteyukiStudio/devops/internal/provider/kubernetes"
 	"github.com/LiteyukiStudio/devops/internal/tasks"
 	"github.com/gin-gonic/gin"
+	"github.com/hibiken/asynq"
 	"gorm.io/gorm"
 )
 
@@ -149,5 +151,5 @@ func (host applicationHost) EnqueueApplicationDelete(ctx context.Context, app mo
 	_, err := host.handlers.taskClient.EnqueueApplicationDelete(ctx, tasks.ApplicationDeletePayload{
 		ApplicationID: app.ID, ProjectID: app.ProjectID, ActorID: actorID, DeleteData: deleteData,
 	})
-	return err == nil
+	return err == nil || errors.Is(err, asynq.ErrDuplicateTask)
 }

internal/api/handlers_core_test.go

@@ -152,7 +152,7 @@ func TestHighGrowthListHandlersCannotBypassPaginationContract(t *testing.T) {
 		t.Run(contract.function, func(t *testing.T) {
 			function := parseAPIFunction(t, contract.file, contract.function)
 			calls := calledFunctions(function.Body)
-			if !calls["paginatedResponse"] {
+			if !calls["paginatedResponse"] && !calls["clusterResourcePaginatedResponse"] {
 				t.Fatalf("%s must return the shared pagination envelope", contract.function)
 			}
 			sortNormalized := calls["paginationFromQueryWithSort"] || calls["normalizeClusterResourceSortBy"] || calls["gitRepositoryPagination"]
@@ -163,7 +163,7 @@ func TestHighGrowthListHandlersCannotBypassPaginationContract(t *testing.T) {
 				calls["buildRunPageQuery"] || calls["containerImagePageQuery"] ||
 				calls["deploymentTargetPageQuery"] || calls["projectPageQuery"] ||
 				calls["ListRepositories"] || calls["SearchPublicRepositories"] ||
-				calls["ListManagedResourcesPage"] || calls["ListManagedResourceEventsPage"]
+				calls["clusterResourcePaginatedResponse"] || calls["ListManagedResourceEventsPage"]
 			if !bounded {
 				t.Fatalf("%s must apply LIMIT/OFFSET or bounded slice pagination", contract.function)
 			}

internal/api/handlers_tasks_test.go

@@ -74,6 +74,19 @@ func TestEnqueueResourceCleanupPassesActor(t *testing.T) {
 	}
 }
 
+func TestEnqueueApplicationDeleteTreatsExistingTaskAsQueued(t *testing.T) {
+	fake := &fakeBuildTaskEnqueuer{applicationDeleteError: asynq.ErrDuplicateTask}
+	h := &Handlers{taskClient: fake}
+
+	app := model.Application{ID: "app_1", ProjectID: "prj_1"}
+	if !((applicationHost{domainHost{handlers: h}}).EnqueueApplicationDelete(context.Background(), app, "usr_operator", false)) {
+		t.Fatal("existing application deletion task must be treated as queued")
+	}
+	if fake.applicationDeletePayload.ApplicationID != app.ID || fake.applicationDeletePayload.ProjectID != app.ProjectID {
+		t.Fatalf("application deletion payload = %#v", fake.applicationDeletePayload)
+	}
+}
+
 func TestRollbackReleaseFromTargetUsesPreviousSuccessfulRelease(t *testing.T) {
 	source := model.Release{
 		ID:            "rel_current",
@@ -205,6 +218,7 @@ type fakeBuildTaskEnqueuer struct {
 	gatewayPayload           tasks.GatewayApplyPayload
 	notificationPayload      tasks.NotificationDeliverPayload
 	applicationDeletePayload tasks.ApplicationDeletePayload
+	applicationDeleteError   error
 	resourceCleanupPayload   tasks.ResourceCleanupPayload
 }
 
@@ -230,7 +244,7 @@ func (f *fakeBuildTaskEnqueuer) EnqueueNotificationDeliver(_ context.Context, pa
 
 func (f *fakeBuildTaskEnqueuer) EnqueueApplicationDelete(_ context.Context, payload tasks.ApplicationDeletePayload) (*asynq.TaskInfo, error) {
 	f.applicationDeletePayload = payload
-	return &asynq.TaskInfo{}, nil
+	return &asynq.TaskInfo{}, f.applicationDeleteError
 }
 
 func (f *fakeBuildTaskEnqueuer) EnqueueResourceCleanup(_ context.Context, payload tasks.ResourceCleanupPayload) (*asynq.TaskInfo, error) {

internal/api/runtimeapi/runtime_cluster_delete_test.go

@@ -0,0 +1,72 @@
+package runtimeapi
+
+import (
+	"testing"
+
+	"github.com/LiteyukiStudio/devops/internal/model"
+	"github.com/LiteyukiStudio/devops/internal/testdb"
+	"gorm.io/gorm"
+)
+
+func TestRuntimeClusterReferenceCountsCoverImplicitTargetsAndProjectVolumes(t *testing.T) {
+	db := testdb.Open(t, testdb.Options{
+		SchemaPrefix: "runtime_cluster_delete_refs_test",
+		Migrate: func(db *gorm.DB) error {
+			return db.AutoMigrate(&model.Project{}, &model.RuntimeCluster{}, &model.DeploymentTarget{}, &model.ProjectVolume{})
+		},
+	})
+	projects := []model.Project{
+		{ID: "prj_one", Identifier: "one", Name: "One"},
+		{ID: "prj_two", Identifier: "two", Name: "Two"},
+		{ID: "prj_three", Identifier: "three", Name: "Three"},
+	}
+	if err := db.Create(&projects).Error; err != nil {
+		t.Fatalf("create projects: %v", err)
+	}
+	defaultCluster := model.RuntimeCluster{ID: "clu_default", Name: "Default", Scope: "global", IsDefault: true, DeleteStatus: "active"}
+	otherCluster := model.RuntimeCluster{ID: "clu_other", Name: "Other", Scope: "global", DeleteStatus: "active"}
+	if err := db.Create(&[]model.RuntimeCluster{defaultCluster, otherCluster}).Error; err != nil {
+		t.Fatalf("create runtime clusters: %v", err)
+	}
+	targets := []model.DeploymentTarget{
+		{ID: "dplt_implicit", ProjectID: "prj_one", ApplicationID: "app_one", Name: "Implicit", Stage: "prod"},
+		{ID: "dplt_default", ProjectID: "prj_two", ApplicationID: "app_two", Name: "Default", Stage: "prod", ClusterID: defaultCluster.ID},
+		{ID: "dplt_other", ProjectID: "prj_three", ApplicationID: "app_three", Name: "Other", Stage: "prod", ClusterID: otherCluster.ID},
+	}
+	if err := db.Create(&targets).Error; err != nil {
+		t.Fatalf("create deployment targets: %v", err)
+	}
+	volumes := []model.ProjectVolume{
+		projectVolumeReferenceFixture("pvol_default", "prj_one", defaultCluster.ID),
+		projectVolumeReferenceFixture("pvol_other", "prj_two", otherCluster.ID),
+	}
+	if err := db.Create(&volumes).Error; err != nil {
+		t.Fatalf("create project volumes: %v", err)
+	}
+
+	targetCount, volumeCount, err := runtimeClusterReferenceCounts(db, defaultCluster)
+	if err != nil {
+		t.Fatalf("default runtime cluster reference counts: %v", err)
+	}
+	if targetCount != 2 || volumeCount != 1 {
+		t.Fatalf("default cluster references = targets:%d volumes:%d, want 2 and 1", targetCount, volumeCount)
+	}
+	targetCount, volumeCount, err = runtimeClusterReferenceCounts(db, otherCluster)
+	if err != nil {
+		t.Fatalf("other runtime cluster reference counts: %v", err)
+	}
+	if targetCount != 1 || volumeCount != 1 {
+		t.Fatalf("other cluster references = targets:%d volumes:%d, want 1 and 1", targetCount, volumeCount)
+	}
+}
+
+func projectVolumeReferenceFixture(id, projectID, clusterID string) model.ProjectVolume {
+	return model.ProjectVolume{
+		ID: id, ProjectID: projectID, DisplayName: id, ClusterID: clusterID,
+		Namespace: "ns-" + projectID, ClaimName: id, OwnershipMode: model.ProjectVolumeOwnershipManaged,
+		SourceKind: model.ProjectVolumeSourceBlank, LifecycleState: model.ProjectVolumeLifecycleReady,
+		CapacityRequest: "1Gi", CapacityBytes: 1024 * 1024 * 1024,
+		AccessMode: model.ProjectVolumeAccessReadWriteOnce, VolumeMode: model.ProjectVolumeModeFilesystem,
+		CreatedBy: "usr_test",
+	}
+}

internal/api/runtimeapi/runtime_cluster_handlers.go

@@ -2,6 +2,7 @@ package runtimeapi
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"strings"
 	"time"
@@ -227,13 +228,17 @@ func (h *Handlers) DeleteRuntimeCluster(ctx *gin.Context) {
 	if !h.canManageScopedResourceByID(ctx, user, cluster.Scope, cluster.OwnerRef, scopedResourceRuntimeCluster, cluster.ID, "无权维护该运行集群") {
 		return
 	}
-	var targetCount int64
-	if err := h.dbFor(ctx).Model(&model.DeploymentTarget{}).Where("cluster_id = ?", cluster.ID).Count(&targetCount).Error; err != nil {
+	targetCount, volumeCount, err := runtimeClusterReferenceCounts(h.dbFor(ctx), cluster)
+	if err != nil {
 		writeError(ctx, http.StatusInternalServerError, err.Error())
 		return
 	}
 	if targetCount > 0 {
-		writeError(ctx, http.StatusConflict, "运行集群仍被部署配置引用，请先迁移或删除相关部署配置")
+		writeErrorCode(ctx, http.StatusConflict, "runtime_cluster.in_use", "runtime cluster is referenced by active deployment targets")
+		return
+	}
+	if volumeCount > 0 {
+		writeErrorCode(ctx, http.StatusConflict, "runtime_cluster.in_use", "runtime cluster is referenced by active project volumes")
 		return
 	}
 	if !h.deleteStatusCanStart(cluster.DeleteStatus) {
@@ -260,6 +265,39 @@ func (h *Handlers) DeleteRuntimeCluster(ctx *gin.Context) {
 	ctx.Status(http.StatusNoContent)
 }
 
+func runtimeClusterReferenceCounts(db *gorm.DB, cluster model.RuntimeCluster) (int64, int64, error) {
+	targetCount, volumeCount, err := runtimecluster.ActiveProjectReferenceCounts(db, cluster.ID)
+	if err != nil {
+		return 0, 0, err
+	}
+	if cluster.Scope == "global" {
+		var defaultCluster model.RuntimeCluster
+		err = runtimecluster.ActiveScope(db).
+			Where("scope = ? and is_default = ?", "global", true).
+			First(&defaultCluster).Error
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			err = runtimecluster.ActiveScope(db).
+				Where("scope = ?", "global").
+				Order("created_at asc").
+				First(&defaultCluster).Error
+		}
+		if err != nil {
+			return 0, 0, err
+		}
+		if defaultCluster.ID == cluster.ID {
+			var implicitTargetCount int64
+			if err := db.Model(&model.DeploymentTarget{}).
+				Joins("join projects on projects.id = deployment_targets.project_id and projects.deleted_at is null").
+				Where("cluster_id = ''").
+				Count(&implicitTargetCount).Error; err != nil {
+				return 0, 0, err
+			}
+			targetCount += implicitTargetCount
+		}
+	}
+	return targetCount, volumeCount, nil
+}
+
 func (h *Handlers) TestRuntimeCluster(ctx *gin.Context) {
 	markLiveObservationResponse(ctx)
 	user, ok := h.currentUser(ctx)

internal/api/runtimeapi/runtime_cluster_resource_handlers.go

@@ -37,10 +37,6 @@ func (h *Handlers) ListRuntimeClusterResources(ctx *gin.Context) {
 		return
 	}
 	pagination := paginationFromQuery(ctx)
-	if pagination.Page != 1 {
-		writeErrorCode(ctx, http.StatusBadRequest, "pagination.cursor_required", "runtime cluster resources only support the first bounded page")
-		return
-	}
 	kubeconfig := h.secrets.ResolveContext(ctx.Request.Context(), cluster.KubeconfigRef)
 	if strings.TrimSpace(kubeconfig) == "" {
 		writeError(ctx, http.StatusBadRequest, "运行集群缺少 kubeconfig，无法读取资源")
@@ -57,7 +53,6 @@ func (h *Handlers) ListRuntimeClusterResources(ctx *gin.Context) {
 		ProjectID:          strings.TrimSpace(ctx.Query("projectId")),
 		ApplicationID:      strings.TrimSpace(ctx.Query("applicationId")),
 		DeploymentTargetID: strings.TrimSpace(ctx.Query("deploymentTargetId")),
-		Limit:              int64(pagination.PageSize),
 	}
 	if !validRuntimeResourceCategory(options.Kind) {
 		writeRuntimeResourceArgumentError(ctx, "cluster.resource_category_invalid", "resourceCategory", runtimeResourceCategories)
@@ -68,12 +63,12 @@ func (h *Handlers) ListRuntimeClusterResources(ctx *gin.Context) {
 	}
 	requestCtx, cancel := context.WithTimeout(ctx.Request.Context(), 10*time.Second)
 	defer cancel()
-	page, err := client.ListManagedResourcesPage(requestCtx, options)
+	items, err := client.ListManagedResources(requestCtx, options)
 	if err != nil {
 		writeError(ctx, http.StatusBadGateway, "集群资源读取失败，请检查集群连接和权限")
 		return
 	}
-	items := h.filterClusterResourceSnapshots(ctx, user, page.Items, visibility, options.ProjectID)
+	items = h.filterClusterResourceSnapshots(ctx, user, items, visibility, options.ProjectID)
 	responses, err := h.clusterResourceResponses(items, ctx.Request.Context())
 	if err != nil {
 		writeError(ctx, http.StatusInternalServerError, err.Error())
@@ -83,13 +78,7 @@ func (h *Handlers) ListRuntimeClusterResources(ctx *gin.Context) {
 		responses = groupWorkloadPodResponses(responses)
 	}
 	pagination.SortBy = normalizeClusterResourceSortBy(pagination.SortBy)
-	sortClusterResourceResponses(responses, pagination)
-	pageItems := paginateSlice(responses, pagination)
-	// A resource category fans out to several Kubernetes kinds, each with an
-	// independent continue token. A numeric global page cannot preserve the
-	// requested cross-kind sort without draining every kind. Keep this endpoint
-	// to one bounded page and do not advertise inaccessible remaining pages.
-	ctx.JSON(http.StatusOK, paginatedResponse(pageItems, int64(len(pageItems)), pagination))
+	ctx.JSON(http.StatusOK, clusterResourcePaginatedResponse(responses, pagination))
 }
 
 func (h *Handlers) GetRuntimeClusterResourceYAML(ctx *gin.Context) {

internal/api/runtimeapi/runtime_cluster_resource_responses.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	transportapi "github.com/LiteyukiStudio/devops/internal/api/transport"
 	"github.com/LiteyukiStudio/devops/internal/model"
 	kubeprovider "github.com/LiteyukiStudio/devops/internal/provider/kubernetes"
 )
@@ -178,6 +179,11 @@ func sortClusterResourceResponses(items []clusterResourceResponse, pagination pa
 	})
 }
 
+func clusterResourcePaginatedResponse(items []clusterResourceResponse, pagination paginationParams) transportapi.PaginatedResponseBody[clusterResourceResponse] {
+	sortClusterResourceResponses(items, pagination)
+	return paginatedResponse(paginateSlice(items, pagination), int64(len(items)), pagination)
+}
+
 func groupWorkloadPodResponses(items []clusterResourceResponse) []clusterResourceResponse {
 	deployments := make([]clusterResourceResponse, 0, len(items))
 	pods := make([]clusterResourceResponse, 0)

internal/api/runtimeapi/runtime_cluster_resource_responses_test.go

@@ -1,6 +1,9 @@
 package runtimeapi
 
-import "testing"
+import (
+	"fmt"
+	"testing"
+)
 
 func TestGroupWorkloadPodResponses(t *testing.T) {
 	items := []clusterResourceResponse{
@@ -27,3 +30,24 @@ func TestGroupWorkloadPodResponses(t *testing.T) {
 		t.Fatalf("expected unmatched pod to stay top-level, got %s/%s", grouped[2].Kind, grouped[2].Name)
 	}
 }
+
+func TestClusterResourcePaginatedResponseUsesCompleteSortedSet(t *testing.T) {
+	filteredItems := make([]clusterResourceResponse, 0, 25)
+	for index := 24; index >= 0; index-- {
+		filteredItems = append(filteredItems, clusterResourceResponse{Name: fmt.Sprintf("resource-%02d", index)})
+	}
+
+	response := clusterResourcePaginatedResponse(filteredItems, paginationParams{
+		Page:      2,
+		PageSize:  10,
+		SortBy:    "name",
+		SortOrder: "asc",
+	})
+
+	if response.Page != 2 || response.Total != 25 || response.TotalPages != 3 {
+		t.Fatalf("pagination = page %d, total %d, totalPages %d", response.Page, response.Total, response.TotalPages)
+	}
+	if len(response.Items) != 10 || response.Items[0].Name != "resource-10" || response.Items[9].Name != "resource-19" {
+		t.Fatalf("second page = %#v", response.Items)
+	}
+}

internal/database/fresh_database_migration_test.go

@@ -21,16 +21,21 @@ import (
 	"gorm.io/gorm/schema"
 )
 
-const baselineMigrationVersion = 103
+const (
+	baselineMigrationVersion = 103
+	latestMigrationVersion   = 104
+)
 
-func TestEmbeddedMigrationsContainSingleBaseline(t *testing.T) {
+func TestEmbeddedMigrationsContainExpectedFiles(t *testing.T) {
 	entries, err := sqlmigrations.FS.ReadDir(".")
 	if err != nil {
 		t.Fatalf("read embedded migrations: %v", err)
 	}
-	baselineFiles := map[string]bool{
-		"000103_baseline.up.sql":   false,
-		"000103_baseline.down.sql": false,
+	migrationFiles := map[string]bool{
+		"000103_baseline.up.sql":                  false,
+		"000103_baseline.down.sql":                false,
+		"000104_bridge_retained_volumes.up.sql":   false,
+		"000104_bridge_retained_volumes.down.sql": false,
 	}
 	migrationCount := 0
 	for _, entry := range entries {
@@ -39,17 +44,17 @@ func TestEmbeddedMigrationsContainSingleBaseline(t *testing.T) {
 			continue
 		}
 		migrationCount++
-		if _, ok := baselineFiles[name]; !ok {
+		if _, ok := migrationFiles[name]; !ok {
 			t.Fatalf("unexpected embedded migration %s", name)
 		}
-		baselineFiles[name] = true
+		migrationFiles[name] = true
 	}
-	if migrationCount != len(baselineFiles) {
-		t.Fatalf("embedded migration file count = %d, want %d", migrationCount, len(baselineFiles))
+	if migrationCount != len(migrationFiles) {
+		t.Fatalf("embedded migration file count = %d, want %d", migrationCount, len(migrationFiles))
 	}
-	for name, found := range baselineFiles {
+	for name, found := range migrationFiles {
 		if !found {
-			t.Fatalf("baseline is missing %s", name)
+			t.Fatalf("embedded migrations are missing %s", name)
 		}
 	}
 }
@@ -195,8 +200,9 @@ WHERE namespace.nspname = current_schema()
 	}
 	assertRunnerMigrationVersion(t, runner, baselineMigrationVersion)
 	if err := MigrateContext(context.Background(), testDB); err != nil {
-		t.Fatalf("repeat migration after baseline bootstrap: %v", err)
+		t.Fatalf("apply migrations after baseline bootstrap: %v", err)
 	}
+	assertRunnerMigrationVersion(t, runner, latestMigrationVersion)
 
 	assertFreshMigrationState(t, testDB)
 	assertStableModelMigrationCoverage(t, testDB)
@@ -262,8 +268,8 @@ func assertFreshMigrationState(t *testing.T, db *gorm.DB) {
 	if migrationState.Dirty {
 		t.Fatalf("fresh database migration is dirty at version %d", migrationState.Version)
 	}
-	if migrationState.Version != baselineMigrationVersion {
-		t.Fatalf("migration version = %d, want %d", migrationState.Version, baselineMigrationVersion)
+	if migrationState.Version != latestMigrationVersion {
+		t.Fatalf("migration version = %d, want %d", migrationState.Version, latestMigrationVersion)
 	}
 	assertSlimmingMigrationRemovals(t, db)
 	assertBaselineConstraintsAndDefaults(t, db)