fix: spread vreplicas across minReplicas pods in scheduler

pkg/scheduler/statefulset/scheduler.go

@@ -115,6 +115,10 @@ type StatefulSetScheduler struct {
 	// replicas is the (cached) number of statefulset replicas.
 	replicas int32
 
+	// minReplicas is the minimum number of replicas for the statefulset.
+	// When set, the scheduler spreads vreplicas across at least this many pods.
+	minReplicas int32
+
 	// isLeader signals whether a given Scheduler instance is leader or not.
 	// The autoscaler is considered the leader when ephemeralLeaderElectionObject is in a
 	// bucket where we've been promoted.
@@ -225,6 +229,7 @@ func newStatefulSetScheduler(ctx context.Context,
 		stateAccessor:        stateAccessor,
 		reserved:             make(map[types.NamespacedName]map[string]int32),
 		autoscaler:           autoscaler,
+		minReplicas:          cfg.MinReplicas,
 	}
 
 	// Monitor our statefulset
@@ -377,34 +382,50 @@ func (s *StatefulSetScheduler) scheduleVPod(ctx context.Context, vpod scheduler.
 	// - allocates as many vreplicas as possible to the same pod(s)
 	// - allocates remaining vreplicas to new pods
 
+	// When minReplicas is set, ensure vreplicas are spread across at least
+	// that many pods by inflating the target to max(requested, minReplicas).
+	// Cap at schedulable pods to avoid requeueing when pods aren't ready yet.
+	target := vpod.GetVReplicas()
+	if s.minReplicas > target {
+		schedulable := int32(len(state.SchedulablePods)) //nolint:gosec // G115: len is bounded by cluster size, cannot overflow int32
+		if s.minReplicas < schedulable {
+			target = s.minReplicas
+		} else {
+			target = schedulable
+		}
+		if target < vpod.GetVReplicas() {
+			target = vpod.GetVReplicas()
+		}
+	}
+
 	// Exact number of vreplicas => do nothing
 	tr := scheduler.GetTotalVReplicas(placements)
-	if tr == vpod.GetVReplicas() {
+	if tr == target {
 		logger.Debug("scheduling succeeded (already scheduled)")
 
 		// Fully placed. Nothing to do
 		return placements, nil
 	}
 
 	// Need less => scale down
-	if tr > vpod.GetVReplicas() {
-		logger.Debugw("scaling down", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()),
+	if tr > target {
+		logger.Debugw("scaling down", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", target),
 			zap.Any("placements", placements),
 			zap.Any("existingPlacements", existingPlacements))
 
-		placements = s.removeReplicas(tr-vpod.GetVReplicas(), placements)
+		placements = s.removeReplicas(tr-target, placements)
 
 		// Do not trigger the autoscaler to avoid unnecessary churn
 
 		return placements, nil
 	}
 
 	// Need more => scale up
-	logger.Debugw("scaling up", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()),
+	logger.Debugw("scaling up", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", target),
 		zap.Any("placements", placements),
 		zap.Any("existingPlacements", existingPlacements))
 
-	placements, left := s.addReplicas(state, reservedByPodName, vpod, vpod.GetVReplicas()-tr, placements)
+	placements, left := s.addReplicas(state, reservedByPodName, vpod, target-tr, placements)
 
 	if left > 0 {
 		// Give time for the autoscaler to do its job

pkg/scheduler/statefulset/scheduler_test.go

@@ -60,6 +60,7 @@ func TestStatefulsetScheduler(t *testing.T) {
 		expectedReserved  map[types.NamespacedName]map[string]int32
 		unschedulablePods sets.Set[int32]
 		capacity          int32
+		minReplicas       int32
 	}{
 		{
 			name:      "no replicas, no vreplicas",
@@ -699,6 +700,76 @@ func TestStatefulsetScheduler(t *testing.T) {
 			},
 			unschedulablePods: sets.New[int32](1),
 		},
+		{
+			name:        "minReplicas=3, 1 vreplica, 3 replicas, spread across all pods",
+			vreplicas:   1,
+			replicas:    int32(3),
+			minReplicas: 3,
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+				{PodName: "statefulset-name-1", VReplicas: 1},
+				{PodName: "statefulset-name-2", VReplicas: 1},
+			},
+		},
+		{
+			name:        "minReplicas=3, 1 vreplica, 1 replica, place what we can",
+			vreplicas:   1,
+			replicas:    int32(1),
+			minReplicas: 3,
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+			},
+		},
+		{
+			name:        "minReplicas=3, 5 vreplicas, 3 replicas, vreplicas > minReplicas uses normal behavior",
+			vreplicas:   5,
+			replicas:    int32(3),
+			minReplicas: 3,
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 2},
+				{PodName: "statefulset-name-1", VReplicas: 2},
+				{PodName: "statefulset-name-2", VReplicas: 1},
+			},
+		},
+		{
+			name:        "minReplicas=0, 1 vreplica, 1 replica, disabled minReplicas uses normal behavior",
+			vreplicas:   1,
+			replicas:    int32(1),
+			minReplicas: 0,
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+			},
+		},
+		{
+			name:        "minReplicas=3, already spread across 3 pods, no change",
+			vreplicas:   1,
+			replicas:    int32(3),
+			minReplicas: 3,
+			placements: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+				{PodName: "statefulset-name-1", VReplicas: 1},
+				{PodName: "statefulset-name-2", VReplicas: 1},
+			},
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+				{PodName: "statefulset-name-1", VReplicas: 1},
+				{PodName: "statefulset-name-2", VReplicas: 1},
+			},
+		},
+		{
+			name:        "minReplicas reduced from 3 to 1, scale down placements",
+			vreplicas:   1,
+			replicas:    int32(3),
+			minReplicas: 1,
+			placements: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+				{PodName: "statefulset-name-1", VReplicas: 1},
+				{PodName: "statefulset-name-2", VReplicas: 1},
+			},
+			expected: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 1},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -753,6 +824,7 @@ func TestStatefulsetScheduler(t *testing.T) {
 				StatefulSetNamespace: testNs,
 				StatefulSetName:      sfsName,
 				VPodLister:           vpodClient.List,
+				MinReplicas:          tc.minReplicas,
 			}
 			s := newStatefulSetScheduler(ctx, cfg, sa, nil)
 			err = s.Promote(reconciler.UniversalBucket(), func(bucket reconciler.Bucket, name types.NamespacedName) {})