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,19 @@ 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, Pending, and Unknown pods are considered live to avoid premature replacement.
+	// Unknown phase (e.g. node partition) should not trigger re-allocation until Kubernetes
+	// declares the pod terminal. Only truly terminal pods (Failed/Succeeded) or deleted pods
+	// should trigger re-allocation.
 	livePodSet := make(map[string]struct{}, len(spec.Pods))
+	// allPodSet tracks all pods that still have an object (including terminal ones).
+	// Used to distinguish "deleted" (object gone) from "terminal" (object exists but Failed/Succeeded).
+	allPodSet := make(map[string]struct{}, len(spec.Pods))
 	for _, p := range spec.Pods {
-		livePodSet[p.Name] = struct{}{}
+		allPodSet[p.Name] = struct{}{}
+		if p.Status.Phase != corev1.PodFailed && p.Status.Phase != corev1.PodSucceeded {
+			livePodSet[p.Name] = struct{}{}
+		}
 	}
 
 	// Fetch pool allocation once and reuse it for both stale-sandbox cleanup and available-pod filtering.
@@ -479,7 +489,7 @@ func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec
 	// handles them without any special-casing outside this function.
 	// Terminating sandboxes are handled inside getSandboxRequest: they receive no new supplement and
 	// all unreleased pods are queued for release.
-	allRequest, err := allocator.getAllRequest(ctx, spec.Sandboxes, podAllocation, livePodSet)
+	allRequest, err := allocator.getAllRequest(ctx, spec.Sandboxes, podAllocation, livePodSet, allPodSet)
 	if err != nil {
 		return nil, err
 	}
@@ -500,13 +510,13 @@ func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec
 // orphan entries for pods in podAllocation whose sandbox is no longer in the sandboxes list
 // (e.g. force-deleted). Orphan entries carry PodSupplement=0 and ToRelease set to the orphan
 // pods so the normal recycle path handles them without special-casing in the caller.
-func (allocator *defaultAllocator) getAllRequest(ctx context.Context, sandboxes []*sandboxv1alpha1.BatchSandbox, podAllocation map[string]string, livePodSet map[string]struct{}) ([]*algorithm.SandboxRequest, error) {
+func (allocator *defaultAllocator) getAllRequest(ctx context.Context, sandboxes []*sandboxv1alpha1.BatchSandbox, podAllocation map[string]string, livePodSet map[string]struct{}, allPodSet map[string]struct{}) ([]*algorithm.SandboxRequest, error) {
 	log := logf.FromContext(ctx)
 	existingSandboxes := make(map[string]struct{}, len(sandboxes))
 	allRequest := make([]*algorithm.SandboxRequest, 0, len(sandboxes))
 	for _, sandbox := range sandboxes {
 		existingSandboxes[sandbox.Name] = struct{}{}
-		request, err := allocator.getSandboxRequest(ctx, sandbox, livePodSet)
+		request, err := allocator.getSandboxRequest(ctx, sandbox, livePodSet, allPodSet)
 		if err != nil {
 			return nil, err
 		}
@@ -529,7 +539,7 @@ func (allocator *defaultAllocator) getAllRequest(ctx context.Context, sandboxes
 	return allRequest, nil
 }
 
-func (allocator *defaultAllocator) getSandboxRequest(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, livePodSet map[string]struct{}) (*algorithm.SandboxRequest, error) {
+func (allocator *defaultAllocator) getSandboxRequest(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, livePodSet map[string]struct{}, allPodSet map[string]struct{}) (*algorithm.SandboxRequest, error) {
 	log := logf.FromContext(ctx)
 	allocated, err := allocator.GetSandboxAllocation(ctx, sandbox)
 	if err != nil {
@@ -545,21 +555,34 @@ 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).
+	// Terminal pods (Failed/Succeeded) that still exist are excluded from liveAllocated
+	// but NOT added to deadPods/ToRelease — they should not go through the recycle path
+	// (e.g. RestartRecycler rejects non-Running pods). Pool scaleDown handles their cleanup.
 	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]
+		_, exists := allPodSet[p]
+		if isReleased && !isAlive {
+			// Released and gone/terminal — no longer occupies a slot.
 			continue
 		}
-		if _, alive := livePodSet[p]; alive {
-			liveAllocated = append(liveAllocated, p)
-		} else {
-			deadPods = append(deadPods, p)
+		if !isReleased && !isAlive {
+			if !exists {
+				// Pod object is completely gone — externally deleted, queue for release.
+				deadPods = append(deadPods, p)
+			}
+			// If exists but terminal (Failed/Succeeded): don't count as live,
+			// don't queue for release. Pool cleanup will handle the terminal pod.
+			continue
 		}
+		liveAllocated = append(liveAllocated, p)
 	}
 	if len(deadPods) > 0 {
 		log.Info("Detected dead allocated pods, queuing for release to trigger re-allocation",
@@ -594,13 +617,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,56 @@ 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{
+				// Terminal pod does not go through ToRelease (avoids restart recycle failure).
+				// Pool scaleDown handles terminal pod cleanup separately.
+				ToAllocate:    map[string][]string{"sbx1": {"pod2"}},
+				ToRelease:     map[string][]string{},
+				PodSupplement: 0,
+			},
+		},
 		{
 			name: "orphan sandbox - pods in store but sandbox no longer in spec",
 			spec: &AllocSpec{