feat(routing): least-busy target selection (least_busy strategy)

crates/aisix-admin/src/openapi.rs

@@ -3736,6 +3736,7 @@ fn add_variant_titles(doc: &mut Value) {
                 "Failover",
                 "Least cost",
                 "Least latency",
+                "Least busy",
             ],
         ),
         (

crates/aisix-core/src/models/routing.rs

@@ -22,6 +22,8 @@
 //! - `least_latency`: fastest target first, by a moving average of recent
 //!   observed upstream latency (time-to-first-token for streaming). Targets
 //!   with no latency samples yet rank first so they get probed.
+//! - `least_busy`: least-loaded target first, by the number of in-flight
+//!   requests currently dispatched to each target.
 //!
 //! See [`RoutingStrategy::is_metric_based`].
 
@@ -48,6 +50,9 @@ pub enum RoutingStrategy {
     /// upstream latency (time-to-first-token for streaming), then fall
     /// forward. Targets with no samples yet rank first so they get probed.
     LeastLatency,
+    /// Rank targets least-loaded-first by the number of in-flight requests
+    /// currently dispatched to each target, then fall forward.
+    LeastBusy,
 }
 
 impl RoutingStrategy {
@@ -58,7 +63,7 @@ impl RoutingStrategy {
     pub fn is_metric_based(&self) -> bool {
         matches!(
             self,
-            RoutingStrategy::LeastCost | RoutingStrategy::LeastLatency
+            RoutingStrategy::LeastCost | RoutingStrategy::LeastLatency | RoutingStrategy::LeastBusy
         )
     }
 }
@@ -266,12 +271,18 @@ mod tests {
         )
         .unwrap();
         assert_eq!(latency.strategy, RoutingStrategy::LeastLatency);
+        let busy: Routing = serde_json::from_str(
+            r#"{"strategy":"least_busy","targets":[{"model":"a"},{"model":"b"}]}"#,
+        )
+        .unwrap();
+        assert_eq!(busy.strategy, RoutingStrategy::LeastBusy);
     }
 
     #[test]
     fn is_metric_based_classification() {
         assert!(RoutingStrategy::LeastCost.is_metric_based());
         assert!(RoutingStrategy::LeastLatency.is_metric_based());
+        assert!(RoutingStrategy::LeastBusy.is_metric_based());
         assert!(!RoutingStrategy::Failover.is_metric_based());
         assert!(!RoutingStrategy::RoundRobin.is_metric_based());
         assert!(!RoutingStrategy::Weighted.is_metric_based());

crates/aisix-proxy/src/chat.rs

@@ -1171,6 +1171,12 @@ async fn dispatch(
         // than N simultaneous streams (#450).
         let post_stream_keys = reservation.keys();
         let stream_concurrency_hold = reservation.into_stream_hold();
+        // least_busy: keep this target counted as in-flight for the stream's
+        // full lifetime. Like `stream_concurrency_hold`, the guard is moved
+        // into the on_complete closure (dropped there), so the count stays
+        // raised until the stream completes or is cancelled — the window a
+        // concurrent routing decision must see this target as loaded.
+        let in_flight_hold = state.runtime_status.begin_in_flight(&winner_target_id);
         // Capture everything the stream-completion callback needs so
         // it can fire `emit_usage_event` once the terminal SSE chunk
         // has yielded its `usage` block. Telemetry emission has to
@@ -1386,6 +1392,8 @@ async fn dispatch(
                 // CompleteOnDrop guard on both paths, so the permit is held
                 // for the stream's full lifetime and never leaked (#450).
                 drop(stream_concurrency_hold);
+                // Same lifetime for the least_busy in-flight count.
+                drop(in_flight_hold);
             },
         );
         let response =
@@ -1703,6 +1711,11 @@ async fn dispatch(
             };
 
             let attempt_started = Instant::now();
+            // least_busy: count this target as in-flight for the upstream
+            // call. The response is fully buffered, so the target is done
+            // once `bridge.chat` returns; the guard drops at the end of this
+            // attempt's scope on both the success-break and failure paths.
+            let _in_flight = state.runtime_status.begin_in_flight(&attempt.id);
             let result = bridge.chat(req, &ctx).await;
             let attempt_latency_ms =
                 attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;

crates/aisix-proxy/src/health.rs

@@ -15,7 +15,8 @@
 
 use dashmap::DashMap;
 use std::sync::atomic::AtomicBool;
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
+use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 use axum::http::header::{HeaderName, HeaderValue, CONTENT_TYPE};
@@ -229,6 +230,11 @@ struct RuntimeEntry {
     /// latency in milliseconds. `None` until the first sample. Drives the
     /// `least_latency` routing strategy; independent of health/cooldown.
     latency_ewma_ms: Option<f64>,
+    /// Number of requests currently in flight to this target. Held in an
+    /// `Arc` so an [`InFlightGuard`] can decrement it after the DashMap lock
+    /// is released (and for the streaming path, after the handler returns).
+    /// Drives the `least_busy` routing strategy.
+    in_flight: Arc<AtomicUsize>,
 }
 
 impl RuntimeEntry {
@@ -347,6 +353,21 @@ pub struct ModelRuntimeStatusTracker {
     entries: DashMap<String, RuntimeEntry>,
 }
 
+/// RAII guard that decrements a target's in-flight counter when dropped.
+/// Created by [`ModelRuntimeStatusTracker::begin_in_flight`] before an
+/// upstream attempt. For the streaming path the guard is moved into the
+/// stream body so the count stays raised until the stream ends or is
+/// cancelled, matching the request's true lifetime.
+pub struct InFlightGuard {
+    counter: Arc<AtomicUsize>,
+}
+
+impl Drop for InFlightGuard {
+    fn drop(&mut self) {
+        self.counter.fetch_sub(1, Ordering::Relaxed);
+    }
+}
+
 impl HealthTracker {
     pub fn new() -> Self {
         Self::default()
@@ -489,6 +510,28 @@ impl ModelRuntimeStatusTracker {
         self.entries.get(model_id).and_then(|e| e.latency_ewma_ms)
     }
 
+    /// Mark one request as in flight to `model_id` and return a guard that
+    /// decrements the count when dropped. Drives the `least_busy` strategy.
+    pub fn begin_in_flight(&self, model_id: &str) -> InFlightGuard {
+        let counter = Arc::clone(
+            &self
+                .entries
+                .entry(model_id.to_string())
+                .or_default()
+                .in_flight,
+        );
+        counter.fetch_add(1, Ordering::Relaxed);
+        InFlightGuard { counter }
+    }
+
+    /// Current in-flight request count for `model_id`.
+    pub fn in_flight(&self, model_id: &str) -> usize {
+        self.entries
+            .get(model_id)
+            .map(|e| e.in_flight.load(Ordering::Relaxed))
+            .unwrap_or(0)
+    }
+
     pub fn status(&self, model_id: &str) -> RuntimeStatusSnapshot {
         self.status_with_stale(model_id, None)
     }

crates/aisix-proxy/src/routing.rs

@@ -21,6 +21,9 @@
 //! Models / runtime state, so `resolve_attempt_models` ranks them instead:
 //! - **least_cost**: cheapest target first, by combined input+output per-1K
 //!   price; targets without a `cost` rank last.
+//! - **least_latency**: fastest target first, by an EWMA of observed upstream
+//!   latency; targets with no samples yet rank first (probe, then exploit).
+//! - **least_busy**: least-loaded target first, by in-flight request count.
 
 use aisix_core::{
     AisixSnapshot, Model, Routing, RoutingStrategy, RoutingTarget, WhenAllUnavailablePolicy,
@@ -138,7 +141,9 @@ impl RoutingRegistry {
             RoutingStrategy::Weighted => weighted_pick(&routing.targets),
             // Metric-ordered strategies never reach here — `pick_targets`
             // short-circuits them before computing a start index.
-            RoutingStrategy::LeastCost | RoutingStrategy::LeastLatency => 0,
+            RoutingStrategy::LeastCost
+            | RoutingStrategy::LeastLatency
+            | RoutingStrategy::LeastBusy => 0,
         }
     }
 
@@ -243,6 +248,9 @@ fn order_attempts_by_metric(
                 latency_key(runtime_status, &a.id).total_cmp(&latency_key(runtime_status, &b.id))
             });
         }
+        RoutingStrategy::LeastBusy => {
+            attempts.sort_by_key(|a| runtime_status.in_flight(&a.id));
+        }
         RoutingStrategy::Failover | RoutingStrategy::RoundRobin | RoutingStrategy::Weighted => {}
     }
 }
@@ -845,6 +853,44 @@ mod tests {
         assert!((t.latency_ewma_ms("m").unwrap() - 130.0).abs() < 1e-9);
     }
 
+    // ── order_attempts_by_metric (least_busy) ─────────────────────
+    #[test]
+    fn least_busy_orders_least_loaded_first() {
+        let t = crate::ModelRuntimeStatusTracker::new();
+        let _b1 = t.begin_in_flight("busy");
+        let _b2 = t.begin_in_flight("busy"); // 2 in-flight
+        let _m1 = t.begin_in_flight("mid"); // 1 in-flight
+                                            // "idle" has 0 in-flight.
+        let mut attempts = vec![am("busy"), am("idle"), am("mid")];
+        order_attempts_by_metric(RoutingStrategy::LeastBusy, &mut attempts, &t);
+        let ids: Vec<&str> = attempts.iter().map(|a| a.id.as_str()).collect();
+        assert_eq!(ids, vec!["idle", "mid", "busy"]);
+    }
+
+    #[test]
+    fn least_busy_cold_start_keeps_declaration_order() {
+        let t = crate::ModelRuntimeStatusTracker::new();
+        // All idle (0 in-flight) → stable sort preserves declaration order.
+        let mut attempts = vec![am("a"), am("b"), am("c")];
+        order_attempts_by_metric(RoutingStrategy::LeastBusy, &mut attempts, &t);
+        let ids: Vec<&str> = attempts.iter().map(|a| a.id.as_str()).collect();
+        assert_eq!(ids, vec!["a", "b", "c"]);
+    }
+
+    #[test]
+    fn in_flight_guard_increments_then_decrements_on_drop() {
+        let t = crate::ModelRuntimeStatusTracker::new();
+        assert_eq!(t.in_flight("m"), 0);
+        let g1 = t.begin_in_flight("m");
+        assert_eq!(t.in_flight("m"), 1);
+        let g2 = t.begin_in_flight("m");
+        assert_eq!(t.in_flight("m"), 2);
+        drop(g1);
+        assert_eq!(t.in_flight("m"), 1);
+        drop(g2);
+        assert_eq!(t.in_flight("m"), 0);
+    }
+
     #[test]
     fn metric_strategy_pick_targets_returns_full_declaration_order() {
         let reg = RoutingRegistry::new();

schemas/resources/model.schema.json

@@ -443,6 +443,13 @@
             "least_latency"
           ],
           "type": "string"
+        },
+        {
+          "description": "Rank targets least-loaded-first by the number of in-flight requests currently dispatched to each target, then fall forward.",
+          "enum": [
+            "least_busy"
+          ],
+          "type": "string"
         }
       ]
     },

schemas/resources/routing.schema.json

@@ -98,6 +98,13 @@
           "enum": [
             "least_latency"
           ]
+        },
+        {
+          "description": "Rank targets least-loaded-first by the number of in-flight requests currently dispatched to each target, then fall forward.",
+          "type": "string",
+          "enum": [
+            "least_busy"
+          ]
         }
       ]
     },