fix(allocator): address CR feedback - exclude released and terminal pods from live allocation

- P1: Released pods no longer count toward liveAllocated, preventing
  stale released entries from blocking subsequent re-allocations.
- P2: Only Running+Ready pods are added to livePodSet, so terminal
  pods (Failed/Evicted) that still have their object present also
  trigger re-allocation.

Author: longsuizhi

kubernetes/internal/controller/allocator.go

@@ -461,9 +461,14 @@ func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec
 	}
 
 	// Build a set of live pool pod names for dead-pod detection during allocation requests.
+	// Running pods (regardless of readiness) are considered live to avoid churn from transient
+	// readiness probe flaps. Only truly terminal pods (Failed/Succeeded) or deleted pods
+	// should trigger re-allocation.
 	livePodSet := make(map[string]struct{}, len(spec.Pods))
 	for _, p := range spec.Pods {
-		livePodSet[p.Name] = struct{}{}
+		if p.Status.Phase == corev1.PodRunning || p.Status.Phase == corev1.PodPending {
+			livePodSet[p.Name] = struct{}{}
+		}
 	}
 
 	// Fetch pool allocation once and reuse it for both stale-sandbox cleanup and available-pod filtering.
@@ -545,21 +550,26 @@ func (allocator *defaultAllocator) getSandboxRequest(ctx context.Context, sandbo
 		releasedSet[r] = struct{}{}
 	}
 
-	// Filter out pods that no longer exist in the pool (e.g. externally deleted).
-	// Dead pods are treated as released so the sandbox can receive replacement allocations.
+	// Filter out pods that no longer exist in the pool (e.g. externally deleted) and
+	// pods that have been released AND are no longer live. Only unreleased live pods
+	// should count toward the effective allocation when computing supplement.
+	// Released pods that are still alive keep their slot to avoid premature supplement
+	// (the recycle flow will eventually remove them from alloc-status).
 	liveAllocated := make([]string, 0, len(allocated))
 	deadPods := make([]string, 0)
 	for _, p := range allocated {
-		if _, exists := releasedSet[p]; exists {
-			// Already released, keep in allocated for bookkeeping consistency.
-			liveAllocated = append(liveAllocated, p)
+		_, isReleased := releasedSet[p]
+		_, isAlive := livePodSet[p]
+		if isReleased && !isAlive {
+			// Released and gone — no longer occupies a slot.
 			continue
 		}
-		if _, alive := livePodSet[p]; alive {
-			liveAllocated = append(liveAllocated, p)
-		} else {
+		if !isReleased && !isAlive {
+			// Not released but gone — externally deleted, needs re-allocation.
 			deadPods = append(deadPods, p)
+			continue
 		}
+		liveAllocated = append(liveAllocated, p)
 	}
 	if len(deadPods) > 0 {
 		log.Info("Detected dead allocated pods, queuing for release to trigger re-allocation",
@@ -594,13 +604,20 @@ func (allocator *defaultAllocator) getSandboxRequest(ctx context.Context, sandbo
 	}
 
 	toRelease := make([]string, 0)
+	toReleaseSet := make(map[string]struct{})
 	for _, r := range release {
 		if _, exists := releasedSet[r]; !exists {
 			toRelease = append(toRelease, r)
+			toReleaseSet[r] = struct{}{}
 		}
 	}
 	// Also queue dead pods for release so their allocation records are cleaned up.
-	toRelease = append(toRelease, deadPods...)
+	// Deduplicate against pods already in alloc-release to avoid duplicate recycle operations.
+	for _, dp := range deadPods {
+		if _, exists := toReleaseSet[dp]; !exists {
+			toRelease = append(toRelease, dp)
+		}
+	}
 
 	replica := int32(0)
 	if sandbox.Spec.Replicas != nil {

kubernetes/internal/controller/allocator_test.go

@@ -259,6 +259,54 @@ func TestSchedule(t *testing.T) {
 				PodSupplement: 0,
 			},
 		},
+		{
+			name: "released pod does not block second re-allocation",
+			spec: &AllocSpec{
+				Pods: []*corev1.Pod{
+					// pod1 was previously allocated and released (recycled).
+					// pod2 was the replacement but is now also deleted.
+					// pod3 is a new available pod.
+					{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}, Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}}},
+				},
+				Pool: &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool1"}},
+				Sandboxes: []*sandboxv1alpha1.BatchSandbox{
+					{ObjectMeta: metav1.ObjectMeta{Name: "sbx1"}, Spec: sandboxv1alpha1.BatchSandboxSpec{Replicas: &replica1}},
+				},
+			},
+			poolAlloc:     &PoolAllocation{PodAllocation: map[string]string{"pod1": "sbx1", "pod2": "sbx1"}},
+			sandboxAllocs: map[string]*SandboxAllocation{"sbx1": {Pods: []string{"pod1", "pod2"}}},
+			releases:      map[string]*AllocationRelease{"sbx1": {Pods: []string{}}},
+			released:      map[string]*AllocationReleased{"sbx1": {Pods: []string{"pod1"}}},
+			wantAction: &algorithm.AllocAction{
+				ToAllocate:    map[string][]string{"sbx1": {"pod3"}},
+				ToRelease:     map[string][]string{"sbx1": {"pod2"}},
+				PodSupplement: 0,
+			},
+		},
+		{
+			name: "terminal pod (Failed) triggers re-allocation",
+			spec: &AllocSpec{
+				Pods: []*corev1.Pod{
+					// pod1 exists but is in Failed state — should not count as live.
+					{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}, Status: corev1.PodStatus{Phase: corev1.PodFailed}},
+					// pod2 is ready and available.
+					{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}, Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}}},
+				},
+				Pool: &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool1"}},
+				Sandboxes: []*sandboxv1alpha1.BatchSandbox{
+					{ObjectMeta: metav1.ObjectMeta{Name: "sbx1"}, Spec: sandboxv1alpha1.BatchSandboxSpec{Replicas: &replica1}},
+				},
+			},
+			poolAlloc:     &PoolAllocation{PodAllocation: map[string]string{"pod1": "sbx1"}},
+			sandboxAllocs: map[string]*SandboxAllocation{"sbx1": {Pods: []string{"pod1"}}},
+			releases:      map[string]*AllocationRelease{"sbx1": {Pods: []string{}}},
+			released:      map[string]*AllocationReleased{"sbx1": {Pods: []string{}}},
+			wantAction: &algorithm.AllocAction{
+				ToAllocate:    map[string][]string{"sbx1": {"pod2"}},
+				ToRelease:     map[string][]string{"sbx1": {"pod1"}},
+				PodSupplement: 0,
+			},
+		},
 		{
 			name: "orphan sandbox - pods in store but sandbox no longer in spec",
 			spec: &AllocSpec{