From 9c899d404a7c4095aa593ecc30c10e63125a1cc4 Mon Sep 17 00:00:00 2001
From: Thanos Makatos <thanos.makatos@nutanix.com>
Date: Thu, 24 Sep 2026 21:11:04 +0000
Subject: [PATCH] Apply scaling decisions in a single step

Currently each instance is evaluated independently. While this can
improve performance it can lead to inaccuracies, so evaluate all of
them together. Since the controller has all the scaling decisions,
apply scale downs first to free up resources.

Signed-off-by: Thanos Makatos <thanos.makatos@nutanix.com>
---
 src/controller.rs  | 42 ++++++++++++++++++----
 src/engines/mod.rs | 90 +++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 124 insertions(+), 8 deletions(-)

diff --git a/src/controller.rs b/src/controller.rs
index 98b4fe4..962f84f 100644
--- a/src/controller.rs
+++ b/src/controller.rs
@@ -568,18 +568,46 @@ impl Controller {
 
         let fleet: Vec<_> = self.instances.values().cloned().collect();
         self.persist_new_classifications(&fleet).await?;
-        let plan = self.engine.evaluate_fleet(&fleet, &context).await;
+        let decisions = self.engine.evaluate_fleet(&fleet, &context).await;
+        let mut planned_ids = HashSet::new();
+        let mut plan = Vec::new();
+        for (sequence, item) in decisions.into_iter().enumerate() {
+            if !planned_ids.insert(item.instance_id.clone()) {
+                tracing::warn!(
+                    target: "controller",
+                    vm = %item.instance_id,
+                    "engine returned duplicate decisions; keeping the first"
+                );
+                continue;
+            }
+            let Some(instance) = self.instances.get(&item.instance_id) else {
+                tracing::warn!(
+                    target: "controller",
+                    vm = %item.instance_id,
+                    "engine returned a decision for an unknown VM"
+                );
+                continue;
+            };
+            let action = item.decision;
+            let Some(target) = action.target() else {
+                continue;
+            };
+            let status = instance.status.read().await;
+            let current = status.thread_count;
+            if target == current {
+                continue;
+            }
+            plan.push((target > current, sequence, item.instance_id, action));
+        }
+        plan.sort_by_key(|(is_up, sequence, _, _)| (*is_up, *sequence));
+
         let cap = self.cfg.max_instances_adjusted_per_poll as usize;
         let mut adjusted = 0usize;
-        for item in plan {
-            let action = item.decision;
+        for (_, _, instance_id, action) in plan {
             if cap > 0 && adjusted >= cap {
                 continue;
             }
-            if self
-                .apply_engine_decision(&item.instance_id, action)
-                .await?
-            {
+            if self.apply_engine_decision(&instance_id, action).await? {
                 adjusted += 1;
             }
         }
diff --git a/src/engines/mod.rs b/src/engines/mod.rs
index 08f8976..0c31ec0 100644
--- a/src/engines/mod.rs
+++ b/src/engines/mod.rs
@@ -203,6 +203,10 @@ pub trait ScalingEngine: Send + Sync {
     async fn evaluate(&self, instance: &Arc<Instance>, context: &EngineTickContext) -> ScaleAction;
 
     /// Evaluate eligible VMs concurrently and collect one coherent plan.
+    ///
+    /// Sticky or backend-disabled VMs remain visible to fleet-aware engines
+    /// that override this method, but the default per-VM implementation does
+    /// not evaluate them.
     async fn evaluate_fleet(
         &self,
         instances: &[Arc<Instance>],
@@ -212,7 +216,9 @@ pub trait ScalingEngine: Send + Sync {
         // to first look at all the instances first and then make decisions
         join_all(instances.iter().map(|instance| async move {
             let status = instance.status.read().await;
-            if status.manual_scaling_sticky || !status.scaling_allowed {
+            let scaling_allowed = status.scaling_allowed;
+            let sticky = status.manual_scaling_sticky;
+            if sticky || !scaling_allowed {
                 return None;
             }
             Some(InstanceDecision::new(
@@ -298,4 +304,86 @@ mod tests {
             Err(other) => panic!("expected NoSuchEngine, got {other}"),
         }
     }
+
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use crate::{
+        backends::BackendClientError,
+        instance::{InstanceClient, ThreadPoolSnapshot},
+    };
+
+    struct SnapshotClient;
+
+    #[async_trait]
+    impl InstanceClient for SnapshotClient {
+        async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> {
+            Ok(())
+        }
+
+        async fn get_thread_pool_snapshot(&self) -> Result<ThreadPoolSnapshot, BackendClientError> {
+            Ok(ThreadPoolSnapshot {
+                thread_count: 1,
+                vcpu_count: 2,
+                perf: None,
+                per_thread_util: Some(0.0),
+            })
+        }
+
+        async fn close(&self) {}
+    }
+
+    struct PerInstanceEngine {
+        calls: AtomicUsize,
+    }
+
+    #[async_trait]
+    impl ScalingEngine for PerInstanceEngine {
+        fn name(&self) -> &'static str {
+            "test"
+        }
+
+        fn dump_config(&self) -> serde_json::Value {
+            serde_json::Value::Null
+        }
+
+        async fn evaluate(
+            &self,
+            _instance: &Arc<Instance>,
+            _context: &EngineTickContext,
+        ) -> ScaleAction {
+            self.calls.fetch_add(1, Ordering::Relaxed);
+            ScaleAction::None
+        }
+    }
+
+    fn test_instance(id: &str) -> Arc<Instance> {
+        Arc::new(Instance::new(
+            id.to_string(),
+            Path::new(""),
+            0,
+            SnapshotClient,
+        ))
+    }
+
+    /// Test that default fleet `evaluate` skips ineligible VMs and only
+    /// plans eligible ones.
+    #[tokio::test]
+    async fn default_fleet_evaluation_skips_ineligible_instances() {
+        let engine = PerInstanceEngine {
+            calls: AtomicUsize::new(0),
+        };
+        let first = test_instance("first");
+        let sticky = test_instance("sticky");
+        sticky.status.write().await.manual_scaling_sticky = true;
+
+        let plan = engine
+            .evaluate_fleet(&[first.clone(), sticky], &EngineTickContext::default())
+            .await;
+
+        assert_eq!(engine.calls.load(Ordering::Relaxed), 1);
+        assert_eq!(
+            plan,
+            vec![InstanceDecision::new(&first.id, ScaleAction::None)]
+        );
+    }
 }