[Misc] subscription: maxSubscriptions amended

cmd/server/internal/handler.go

@@ -122,6 +122,20 @@ type SubscriptionHandler struct {
 	Clientset           versioned.Interface
 	KubeClientset       kubernetes.Interface
 	httpClientGenerator httpClientGenerator
+	// appLocks serializes the subscription create path per application (keyed by appIdHash) so that
+	// the MaxSubscriptions limit check and the subsequent create cannot interleave for concurrent
+	// requests. See lockApp and the note on checkSubscriptionLimit for the cross-replica caveat.
+	appLocks sync.Map // appIdHash -> *sync.Mutex
+}
+
+// lockApp lazily creates and acquires the per-application mutex identified by appIdHash and returns
+// the corresponding unlock function. Different applications never block each other; provisioning for
+// a single application is strictly serialized within this process.
+func (s *SubscriptionHandler) lockApp(appIdHash string) func() {
+	actual, _ := s.appLocks.LoadOrStore(appIdHash, &sync.Mutex{})
+	mu := actual.(*sync.Mutex)
+	mu.Lock()
+	return mu.Unlock
 }
 
 type callbackResponse struct {
@@ -185,8 +199,15 @@ func (s *SubscriptionHandler) CreateTenant(reqInfo *RequestInfo) *Result {
 	// Check if a Subscription resource already exists for this payload
 	sub := s.getSubscriptionByAppIdentifier(reqInfo.payload.providerSubaccountId, reqInfo.payload.appName, reqInfo.payload.tenantId, subPro.Namespace, TenantProvisioning).Subscription
 
-	// If the resource doesn't exist, we'll create it
 	if sub == nil {
+		// The create path (limit check + create) is serialized per application so that concurrent
+		// requests cannot each pass the MaxSubscriptions check and then create, overshooting the
+		// limit (TOCTOU). The lock is scoped to this branch so the common update path below is not
+		// serialized.
+		appIdHash := sha1Sum(reqInfo.payload.providerSubaccountId, reqInfo.payload.appName)
+		unlock := s.lockApp(appIdHash)
+		defer unlock()
+
 		limitExceeded, checkErr := s.checkSubscriptionLimit(reqInfo, subPro)
 		if checkErr != nil {
 			util.LogError(checkErr, "Error during subscription limit check", TenantProvisioning, subPro, nil)
@@ -237,6 +258,14 @@ func getMessage(isCreated, isUpdated bool) string {
 // checks whether a new subscription would exceed the MaxSubscriptions limit configured
 // on the SubscriptionProvider. It returns (true, nil) when the limit is reached, (false, nil) when no limit applies,
 // and (false, err) when the check itself fails (in which case the caller should reject the request with a 5xx status).
+//
+// Concurrency: the List it performs is a strongly-consistent live read, so once the create path is
+// serialized per application by CreateTenant (via lockApp) the count reflects all prior creates and
+// the limit is enforced reliably. That serialization is in-process only: it guarantees the limit
+// within a single subscription-server pod (the default, subscriptionServer.replicas: 1). With
+// multiple replicas, requests handled by different pods can still interleave List/Create and
+// overshoot the limit; strict cross-replica enforcement would require a distributed lock (e.g. a
+// coordination.k8s.io Lease per application) and is intentionally out of scope here.
 func (s *SubscriptionHandler) checkSubscriptionLimit(reqInfo *RequestInfo, subPro *v1alpha2.SubscriptionProvider) (bool, error) {
 	if subPro.Spec.MaxSubscriptions == nil {
 		return false, nil // no limit configured

cmd/server/internal/handler_test.go

@@ -18,6 +18,8 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -1328,3 +1330,92 @@ func TestCheckSubscriptionLimit_APIError(t *testing.T) {
 		t.Errorf("expected HTTP %d, got %d", http.StatusInternalServerError, res.Code)
 	}
 }
+
+// TestCheckSubscriptionLimit_ConcurrentRequests verifies that the MaxSubscriptions limit is enforced
+// when many provisioning requests for distinct tenants of the same app arrive concurrently. Without
+// per-app serialization of the check->create path, concurrent requests could each pass the limit
+// check and then create, overshooting the limit (TOCTOU). Run with -race.
+func TestCheckSubscriptionLimit_ConcurrentRequests(t *testing.T) {
+	const (
+		maxSubs     = int32(20)
+		concurrency = 50
+	)
+	subPro := createSubscriptionProvider(SaaS)
+	subMax := maxSubs
+	subPro.Spec.MaxSubscriptions = &subMax
+
+	client, tokenString, err := SetupValidTokenAndIssuerForSubscriptionTests("appname!b14")
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Start with zero subscriptions so exactly maxSubs of the concurrent requests should succeed.
+	fakeClient := fake.NewSimpleClientset(subPro)
+	// The fake clientset does not honor metav1.GenerateName, so concurrent creates would otherwise
+	// all get an empty name and collide. Emulate the API server by assigning a unique name, then fall
+	// through to the default tracker to actually store the object.
+	var nameCounter int64
+	fakeClient.PrependReactor("create", "subscriptions", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		sub := action.(k8stesting.CreateAction).GetObject().(*v1alpha2.Subscription)
+		if sub.Name == "" && sub.GenerateName != "" {
+			sub.Name = fmt.Sprintf("%s%d", sub.GenerateName, atomic.AddInt64(&nameCounter, 1))
+		}
+		return false, nil, nil
+	})
+	subHandler := NewSubscriptionHandler(fakeClient, k8sfake.NewSimpleClientset(createSecrets()...))
+	subHandler.httpClientGenerator = &httpTestClientGenerator{client: client}
+
+	var accepted, rejected int32
+	var wg sync.WaitGroup
+	for i := 0; i < concurrency; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			// Distinct tenant (and guid) per request so each takes the create path, not the update path.
+			tid := fmt.Sprintf("%s-%d", tenantId, i)
+			guid := fmt.Sprintf("%s-%d", subscriptionGUID, i)
+			body := `{"subscriptionAppName":"` + appName + `","globalAccountGUID":"` + globalAccountId + `","providerSubaccountId":"` + providerSubaccountId + `","subscriptionGUID":"` + guid + `","subscribedTenantId":"` + tid + `","subscribedSubdomain":"` + subDomain + `"}`
+			res := httptest.NewRecorder()
+			req := httptest.NewRequest(http.MethodPut, RequestPath, strings.NewReader(body))
+			req.Header.Set("Authorization", "Bearer "+tokenString)
+			subHandler.HandleSaaSRequest(res, req)
+
+			switch res.Code {
+			case http.StatusAccepted:
+				atomic.AddInt32(&accepted, 1)
+			case http.StatusUnprocessableEntity:
+				atomic.AddInt32(&rejected, 1)
+			default:
+				t.Errorf("unexpected HTTP status %d", res.Code)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	// The number of subscriptions actually created must never exceed the configured limit.
+	subList, err := subHandler.Clientset.SmeV1alpha2().Subscriptions(v1.NamespaceDefault).List(
+		context.TODO(), v1.ListOptions{LabelSelector: LabelAppIdHash + "=" + sha1Sum(providerSubaccountId, appName)},
+	)
+	if err != nil {
+		t.Fatal("unexpected error listing subscriptions:", err)
+	}
+	created := 0
+	for i := range subList.Items {
+		if subList.Items[i].DeletionTimestamp == nil {
+			created++
+		}
+	}
+	if int32(created) > maxSubs {
+		t.Errorf("subscription count %d exceeds MaxSubscriptions %d", created, maxSubs)
+	}
+	// Starting from zero, exactly maxSubs requests should have been accepted and the rest rejected.
+	if accepted != maxSubs {
+		t.Errorf("expected %d accepted requests, got %d", maxSubs, accepted)
+	}
+	if int32(created) != maxSubs {
+		t.Errorf("expected %d subscriptions created, got %d", maxSubs, created)
+	}
+	if rejected != concurrency-maxSubs {
+		t.Errorf("expected %d rejected requests, got %d", concurrency-maxSubs, rejected)
+	}
+}