fix(scheduler): filter unreclaimable reclaim victims

Signed-off-by: Erez Freiberger <enoodle@gmail.com>

Author: enoodle

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 ### Changed
 
 ### Fixed
+- Fixed reclaim abandoning valid over-quota victims when an unrelated under-deserved queue appeared earlier in victim ordering. [#1750](https://github.com/kai-scheduler/KAI-Scheduler/issues/1750)
 
 ## [v0.16.0] - 2026-06-24
 

pkg/scheduler/actions/reclaim/reclaim_anchor_victim_test.go

@@ -112,6 +112,13 @@ func TestReclaimAnchorVictim(t *testing.T) {
 				DontValidateGPUGroup: true,
 			},
 		},
+		Mocks: &test_utils.TestMock{
+			CacheRequirements: &test_utils.CacheMocking{
+				NumberOfCacheBinds:      5,
+				NumberOfCacheEvictions:  2,
+				NumberOfPipelineActions: 2,
+			},
+		},
 	}
 
 	ssn := test_utils.BuildSession(topology, controller)

pkg/scheduler/plugins/proportion/proportion.go

@@ -105,6 +105,7 @@ func (pp *proportionPlugin) OnSessionOpen(ssn *framework.Session) {
 	capacityPolicy := cp.New(pp.queues, pp.minNodeGPUMemory)
 	ssn.AddQueueOrderFn(pp.queueOrder)
 	ssn.AddCanReclaimResourcesFn(pp.CanReclaimResourcesFn)
+	ssn.AddReclaimVictimFilterFn(pp.reclaimVictimFilterFn)
 	ssn.AddReclaimScenarioValidatorFn(pp.reclaimableFn)
 	ssn.AddOnJobSolutionStartFn(pp.OnJobSolutionStartFn)
 	ssn.AddIsNonPreemptibleJobOverQueueQuotaFns(capacityPolicy.IsNonPreemptibleJobOverQuota)
@@ -140,6 +141,13 @@ func (pp *proportionPlugin) CanReclaimResourcesFn(reclaimer *podgroup_info.PodGr
 	return pp.reclaimablePlugin.CanReclaimResources(pp.queues, reclaimerInfo)
 }
 
+func (pp *proportionPlugin) reclaimVictimFilterFn(
+	reclaimer *podgroup_info.PodGroupInfo, victim *podgroup_info.PodGroupInfo,
+) bool {
+	reclaimerInfo := pp.buildReclaimerInfo(reclaimer, pp.minNodeGPUMemory)
+	return pp.reclaimablePlugin.FilterVictim(pp.queues, reclaimerInfo, victim.Queue)
+}
+
 func (pp *proportionPlugin) reclaimableFn(
 	scenario api.ScenarioInfo,
 ) bool {

pkg/scheduler/plugins/proportion/reclaimable/filter_victims.go

@@ -0,0 +1,65 @@
+// Copyright 2025 NVIDIA CORPORATION
+// SPDX-License-Identifier: Apache-2.0
+
+package reclaimable
+
+import (
+	commonconstants "github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/resource_info"
+	rs "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/proportion/resource_share"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/proportion/utils"
+)
+
+// FilterVictim removes victims that cannot be reclaimed by the deserved-quota strategy.
+func (r *Reclaimable) FilterVictim(
+	queues map[common_info.QueueID]*rs.QueueAttributes,
+	reclaimer *ReclaimerInfo,
+	reclaimeeQueueID common_info.QueueID,
+) bool {
+	if reclaimer == nil {
+		return true
+	}
+
+	reclaimerQueue, reclaimeeQueue := r.getLeveledQueues(queues, reclaimer.Queue, reclaimeeQueueID)
+	if reclaimerQueue == nil || reclaimeeQueue == nil {
+		return true
+	}
+
+	if !canUseGuaranteeDeservedQuotaStrategy(reclaimer, reclaimerQueue) {
+		return true
+	}
+
+	return canBeDeservedQuotaReclaimCandidate(reclaimer, reclaimeeQueue)
+}
+
+func canUseGuaranteeDeservedQuotaStrategy(
+	reclaimer *ReclaimerInfo, reclaimerQueue *rs.QueueAttributes,
+) bool {
+	allocatedWithReclaimer := reclaimerQueue.GetAllocatedShare()
+	allocatedWithReclaimer.Add(utils.QuantifyVector(reclaimer.RequiredResources, reclaimer.VectorMap))
+	return allocatedWithReclaimer.LessEqual(reclaimerQueue.GetDeservedShare())
+}
+
+func canBeDeservedQuotaReclaimCandidate(
+	reclaimer *ReclaimerInfo, reclaimeeQueue *rs.QueueAttributes,
+) bool {
+	allocated := reclaimeeQueue.GetAllocatedShare()
+	deserved := reclaimeeQueue.GetDeservedShare()
+	involvedResources := getInvolvedResourcesNames([]resource_info.ResourceVector{reclaimer.RequiredResources}, reclaimer.VectorMap)
+
+	hasUnderDeservedResource := false
+	for resource := range involvedResources {
+		if deserved[resource] == commonconstants.UnlimitedResourceQuantity {
+			continue
+		}
+		if allocated[resource] > deserved[resource] {
+			return true
+		}
+		if allocated[resource] < deserved[resource] {
+			hasUnderDeservedResource = true
+		}
+	}
+
+	return !hasUnderDeservedResource
+}

pkg/scheduler/plugins/proportion/reclaimable/reclaimable_test.go

@@ -1134,6 +1134,45 @@ var _ = Describe("Reclaimable - Multiple hierarchy levels", func() {
 	})
 })
 
+var _ = Describe("FilterVictim", func() {
+	reclaimerInfo := &ReclaimerInfo{
+		Queue:             "reclaimer",
+		RequiredResources: resource_info.NewResource(0, 0, 2).ToVector(testVectorMap),
+		VectorMap:         testVectorMap,
+		IsPreemptable:     true,
+	}
+
+	It("filters victims whose leveled queue is strictly under deserved quota", func() {
+		reclaimable := New(1)
+		queues := buildQueues(map[common_info.QueueID]queuesTestData{
+			"reclaimer": {deserved: 4, fairShare: 4, allocated: 0},
+			"victim":    {deserved: 4, fairShare: 4, allocated: 2},
+		})
+
+		Expect(reclaimable.FilterVictim(queues, reclaimerInfo, "victim")).To(BeFalse())
+	})
+
+	It("keeps victims whose leveled queue is over deserved quota", func() {
+		reclaimable := New(1)
+		queues := buildQueues(map[common_info.QueueID]queuesTestData{
+			"reclaimer": {deserved: 4, fairShare: 4, allocated: 0},
+			"victim":    {deserved: 0, fairShare: 4, allocated: 4},
+		})
+
+		Expect(reclaimable.FilterVictim(queues, reclaimerInfo, "victim")).To(BeTrue())
+	})
+
+	It("keeps victims exactly at deserved quota for consolidation", func() {
+		reclaimable := New(1)
+		queues := buildQueues(map[common_info.QueueID]queuesTestData{
+			"reclaimer": {deserved: 4, fairShare: 4, allocated: 0},
+			"victim":    {deserved: 2, fairShare: 4, allocated: 2},
+		})
+
+		Expect(reclaimable.FilterVictim(queues, reclaimerInfo, "victim")).To(BeTrue())
+	})
+})
+
 func buildQueues(queuesData map[common_info.QueueID]queuesTestData) map[common_info.QueueID]*rs.QueueAttributes {
 	queues := map[common_info.QueueID]*rs.QueueAttributes{}
 	for name, queueData := range queuesData {