fix(account): harden PAYG debt recovery (#7089)

* fix(account): recover debt status after balance normalization * fix(account): retry debt refresh when user lock is busy * fix(account): verify workspace subscriptions during debt recovery
2026-08-29 01:39:49 +08:00 · 2026-07-24 11:35:58 +08:00
parent c0034114f5
commit b38d2ae17d
4 changed files with 261 additions and 11 deletions
@@ -716,9 +716,6 @@ func (r *DebtReconciler) refreshDebtStatus(userUID uuid.UUID, skipSendMsg bool)
 	if account == nil {
 		return fmt.Errorf("account %s not found", userUID)
 	}
-	if account.DeductionBalance == 0 {
-		return nil
-	}
 	debt := types.Debt{}
 	err = r.AccountV2.GetGlobalDB().
 		Model(&types.Debt{}).
@@ -731,16 +728,20 @@ func (r *DebtReconciler) refreshDebtStatus(userUID uuid.UUID, skipSendMsg bool)
 	if errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil
 	}
-	isBasicUser := account.Balance <= 10*BaseUnit
-	oweamount := account.Balance - account.DeductionBalance + account.UsableCredits
-	// update interval seconds
-	updateIntervalSeconds := time.Now().UTC().Unix() - debt.UpdatedAt.UTC().Unix()
 	lastStatus := debt.AccountDebtStatus
 	update := false
 	if lastStatus == "" {
 		lastStatus = types.NormalPeriod
 		update = true
 	}
+	// A user can still need debt recovery after ResumeBalance has normalized deduction_balance.
+	if account.DeductionBalance == 0 && !types.ContainDebtStatus(types.DebtStates, lastStatus) {
+		return nil
+	}
+	isBasicUser := account.Balance <= 10*BaseUnit
+	oweamount := account.Balance - account.DeductionBalance + account.UsableCredits
+	// update interval seconds
+	updateIntervalSeconds := time.Now().UTC().Unix() - debt.UpdatedAt.UTC().Unix()
 	currentStatusRaw, err := r.DetermineCurrentStatus(
 		oweamount,
 		account.UserUID,
@@ -1180,6 +1181,7 @@ func (r *DebtReconciler) processUsersInParallel(users []uuid.UUID) {
 			if !mutex.TryLock() {
 				// r.Logger.V(1).Info("user debt processing skipped due to existing lock",
 				//	"userUID", u)
+				r.failedUserLocks.Store(u, 0)
 				return
 			}
 			defer mutex.Unlock()
@@ -67,9 +67,10 @@ func adminFlushDebtResourceStatus(req *helper.AdminFlushDebtResourceStatusReq) e
 	if owner == "" {
 		return nil
 	}
-	namespaces, err := getOwnNsListWithCltWithOutWorkspaceSubscription(
+	namespaces, err := getOwnNsListWithCltForDebtFlush(
 		dao.K8sManager.GetClient(),
 		owner,
+		isDebtRecoveryTransition(req.LastDebtStatus, req.CurrentDebtStatus),
 	)
 	if err != nil {
 		return fmt.Errorf("get own namespace list failed: %w", err)
@@ -80,6 +81,11 @@ func adminFlushDebtResourceStatus(req *helper.AdminFlushDebtResourceStatusReq) e
 	return nil
 }

+func isDebtRecoveryTransition(lastStatus, currentStatus types.DebtStatusType) bool {
+	return types.ContainDebtStatus(types.DebtStates, lastStatus) &&
+		types.ContainDebtStatus(types.NonDebtStates, currentStatus)
+}
+
 func flushUserDebtResourceStatus(
 	req *helper.AdminFlushDebtResourceStatusReq,
 	clt client.Client,
@@ -0,0 +1,181 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/labring/sealos/controllers/pkg/types"
+	"github.com/labring/sealos/service/account/dao"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestGetOwnNsListWithCltForDebtFlush_IncludesDebtSuspendedNamespacesOnRecovery(t *testing.T) {
+	const owner = "test-owner"
+
+	setWorkspaceSubscriptionExistsForTest(t, func(workspace string) (bool, error) {
+		return false, nil
+	})
+
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add core scheme: %v", err)
+	}
+
+	clt := clientfake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(
+			newTestNamespace("payg", owner, nil),
+			newTestNamespace("payg-suspended-with-sub-annotation", owner, map[string]string{
+				types.WorkspaceSubscriptionStatusAnnoKey: types.NormalDebtNamespaceAnnoStatus,
+				types.DebtNamespaceAnnoStatusKey:         types.SuspendCompletedDebtNamespaceAnnoStatus,
+			}),
+			newTestNamespace("subscription-active", owner, map[string]string{
+				types.WorkspaceSubscriptionStatusAnnoKey: types.NormalDebtNamespaceAnnoStatus,
+			}),
+			newTestNamespace("other-owner", "other", nil),
+		).
+		Build()
+
+	ordinaryNamespaces, err := getOwnNsListWithCltForDebtFlush(clt, owner, false)
+	if err != nil {
+		t.Fatalf("get ordinary namespaces: %v", err)
+	}
+	assertStringSet(t, ordinaryNamespaces, []string{"payg"})
+
+	recoveryNamespaces, err := getOwnNsListWithCltForDebtFlush(clt, owner, true)
+	if err != nil {
+		t.Fatalf("get recovery namespaces: %v", err)
+	}
+	assertStringSet(t, recoveryNamespaces, []string{"payg", "payg-suspended-with-sub-annotation"})
+
+	var ns corev1.Namespace
+	if err := clt.Get(context.Background(), client.ObjectKey{Name: "payg-suspended-with-sub-annotation"}, &ns); err != nil {
+		t.Fatalf("get suspended namespace: %v", err)
+	}
+	skip, err := shouldSkipWorkspaceSubscriptionNamespace(ns.Name, ns.Annotations, true)
+	if err != nil {
+		t.Fatalf("check workspace subscription namespace: %v", err)
+	}
+	if skip {
+		t.Fatal("debt-suspended namespace should be included during recovery")
+	}
+}
+
+func TestGetOwnNsListWithCltForDebtFlush_SkipsExistingSubscription(t *testing.T) {
+	const owner = "test-owner"
+
+	setWorkspaceSubscriptionExistsForTest(t, func(workspace string) (bool, error) {
+		return workspace == "subscription-suspended", nil
+	})
+
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add core scheme: %v", err)
+	}
+
+	clt := clientfake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(
+			newTestNamespace("payg", owner, nil),
+			newTestNamespace("subscription-suspended", owner, map[string]string{
+				types.WorkspaceSubscriptionStatusAnnoKey: types.NormalDebtNamespaceAnnoStatus,
+				types.DebtNamespaceAnnoStatusKey:         types.SuspendCompletedDebtNamespaceAnnoStatus,
+			}),
+		).
+		Build()
+
+	recoveryNamespaces, err := getOwnNsListWithCltForDebtFlush(clt, owner, true)
+	if err != nil {
+		t.Fatalf("get recovery namespaces: %v", err)
+	}
+	assertStringSet(t, recoveryNamespaces, []string{"payg"})
+}
+
+func TestGetOwnNsListWithCltForDebtFlush_ReturnsWorkspaceSubscriptionLookupError(t *testing.T) {
+	const owner = "test-owner"
+
+	setWorkspaceSubscriptionExistsForTest(t, func(workspace string) (bool, error) {
+		return false, errors.New("lookup failed")
+	})
+
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add core scheme: %v", err)
+	}
+
+	clt := clientfake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(
+			newTestNamespace("payg-suspended-with-sub-annotation", owner, map[string]string{
+				types.WorkspaceSubscriptionStatusAnnoKey: types.NormalDebtNamespaceAnnoStatus,
+				types.DebtNamespaceAnnoStatusKey:         types.SuspendCompletedDebtNamespaceAnnoStatus,
+			}),
+		).
+		Build()
+
+	_, err := getOwnNsListWithCltForDebtFlush(clt, owner, true)
+	if err == nil {
+		t.Fatal("expected workspace subscription lookup error")
+	}
+	if !strings.Contains(err.Error(), "lookup failed") {
+		t.Fatalf("expected lookup error, got %v", err)
+	}
+}
+
+func TestIsDebtRecoveryTransition(t *testing.T) {
+	if !isDebtRecoveryTransition(types.DebtPeriod, types.NormalPeriod) {
+		t.Fatal("debt to normal should be a recovery transition")
+	}
+	if !isDebtRecoveryTransition(types.DebtDeletionPeriod, types.CriticalBalancePeriod) {
+		t.Fatal("debt deletion to critical balance should be a recovery transition")
+	}
+	if isDebtRecoveryTransition(types.NormalPeriod, types.DebtPeriod) {
+		t.Fatal("normal to debt should not be a recovery transition")
+	}
+	if isDebtRecoveryTransition(types.DebtPeriod, types.DebtDeletionPeriod) {
+		t.Fatal("debt to debt should not be a recovery transition")
+	}
+}
+
+func setWorkspaceSubscriptionExistsForTest(t *testing.T, fn func(workspace string) (bool, error)) {
+	t.Helper()
+	old := workspaceSubscriptionExists
+	workspaceSubscriptionExists = fn
+	t.Cleanup(func() {
+		workspaceSubscriptionExists = old
+	})
+}
+
+func newTestNamespace(name, owner string, annotations map[string]string) *corev1.Namespace {
+	return &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+			Labels: map[string]string{
+				dao.UserOwnerLabel: owner,
+			},
+			Annotations: annotations,
+		},
+	}
+}
+
+func assertStringSet(t *testing.T, got, want []string) {
+	t.Helper()
+	gotSet := make(map[string]struct{}, len(got))
+	for _, item := range got {
+		gotSet[item] = struct{}{}
+	}
+	if len(gotSet) != len(want) {
+		t.Fatalf("got namespaces %v, want %v", got, want)
+	}
+	for _, item := range want {
+		if _, ok := gotSet[item]; !ok {
+			t.Fatalf("got namespaces %v, want %v", got, want)
+		}
+	}
+}
@@ -327,9 +327,27 @@ func getOwnNsListWithClt(clt client.Client, user string) ([]string, error) {
 	return nsListStr, nil
 }

-func getOwnNsListWithCltWithOutWorkspaceSubscription(
+var workspaceSubscriptionExists = func(workspace string) (bool, error) {
+	if dao.DBClient == nil {
+		return false, errors.New("db client is nil")
+	}
+	subscription, err := dao.DBClient.GetWorkspaceSubscription(
+		workspace,
+		dao.DBClient.GetLocalRegion().Domain,
+	)
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("get workspace subscription: %w", err)
+	}
+	return subscription != nil, nil
+}
+
+func getOwnNsListWithCltForDebtFlush(
 	clt client.Client,
 	user string,
+	includeDebtSuspendedNamespaces bool,
 ) ([]string, error) {
 	if user == "" {
 		return nil, errors.New("user is empty")
@@ -344,8 +362,19 @@ func getOwnNsListWithCltWithOutWorkspaceSubscription(
 		if nsList.Items[i].Status.Phase == corev1.NamespaceTerminating {
 			continue
 		}
-		if nsList.Items[i].Annotations != nil &&
-			nsList.Items[i].Annotations[types.WorkspaceSubscriptionStatusAnnoKey] != "" {
+		skipWorkspaceSubscription, err := shouldSkipWorkspaceSubscriptionNamespace(
+			nsList.Items[i].Name,
+			nsList.Items[i].Annotations,
+			includeDebtSuspendedNamespaces,
+		)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"check workspace subscription namespace %s: %w",
+				nsList.Items[i].Name,
+				err,
+			)
+		}
+		if skipWorkspaceSubscription {
 			continue
 		}
 		nsListStr = append(nsListStr, nsList.Items[i].Name)
@@ -353,6 +382,38 @@ func getOwnNsListWithCltWithOutWorkspaceSubscription(
 	return nsListStr, nil
 }

+func shouldSkipWorkspaceSubscriptionNamespace(
+	namespace string,
+	annotations map[string]string,
+	includeDebtSuspendedNamespaces bool,
+) (bool, error) {
+	if annotations == nil || annotations[types.WorkspaceSubscriptionStatusAnnoKey] == "" {
+		return false, nil
+	}
+	if !includeDebtSuspendedNamespaces ||
+		!isDebtSuspendedNamespaceStatus(annotations[types.DebtNamespaceAnnoStatusKey]) {
+		return true, nil
+	}
+	exists, err := workspaceSubscriptionExists(namespace)
+	if err != nil {
+		return true, err
+	}
+	return exists, nil
+}
+
+func isDebtSuspendedNamespaceStatus(status string) bool {
+	switch status {
+	case types.SuspendDebtNamespaceAnnoStatus,
+		types.SuspendCompletedDebtNamespaceAnnoStatus,
+		types.TerminateSuspendDebtNamespaceAnnoStatus,
+		types.TerminateSuspendCompletedDebtNamespaceAnnoStatus,
+		types.FinalDeletionDebtNamespaceAnnoStatus:
+		return true
+	default:
+		return false
+	}
+}
+
 func getDefaultResourceQuota(ns, name string, hard corev1.ResourceList) *corev1.ResourceQuota {
 	return &corev1.ResourceQuota{
 		ObjectMeta: metav1.ObjectMeta{