fix(stargate): expire reservations after backend RTT

docs/observability/metrics/llm-request-router/metrics.md

@@ -38,3 +38,5 @@ unbounded request fields as metric labels.
 | `stargate_proxy_duration_seconds` | Histogram | `llm-request-router:9090/metrics` | `routing_key`, `model`, `inference_server_id` | Time to first byte from upstream in seconds. |
 | `stargate_routing_duration_seconds` | Histogram | `llm-request-router:9090/metrics` | `routing_key`, `model` | Load-balancer decision time in seconds. |
 | `stargate_active_inference_servers` | Gauge | `llm-request-router:9090/metrics` | `routing_key`, `model` | Currently routable inference servers for a routing target. |
+| `stargate_routing_reservations_active` | Gauge | `llm-request-router:9090/metrics` | `routing_key`, `model` | Local routing reservations not yet released or pruned. Expiry is lazy: an expired reservation stays counted until the next snapshot read or registration update for its cluster prunes it. Removing the backend also drops its reservations. |
+| `stargate_routing_reservation_ttl_seconds` | Histogram | `llm-request-router:9090/metrics` | `routing_key`, `model` | RTT-based reservation TTL in seconds after applying the configured bounds. |

src/libraries/rust/stargate/crates/stargate/src/http_proxy.rs

@@ -33,6 +33,8 @@ use crate::tunnel::{QuicHttpProxy, QuicTunnelConfig};
 mod attempt;
 mod diagnostics;
 mod request;
+#[cfg(test)]
+mod reservation_tests;
 mod retry;
 mod routing;
 mod run;
@@ -120,6 +122,8 @@ pub struct ProxyAppState {
     pub readiness: ReadinessState,
     pub lb_router: Arc<LoadBalancerRouter>,
     pub metrics: Arc<StargateMetrics>,
+    pub routing_reservation_ttl_min: std::time::Duration,
+    pub routing_reservation_ttl_max: std::time::Duration,
     pub retry: ProxyRetryConfig,
     pub debug_config: DebugConfig,
 }
@@ -247,7 +251,9 @@ mod test_support {
     use crate::routing_state::StargateState;
     use crate::tunnel::{QuicHttpProxy, QuicTunnelConfig};
 
-    use super::{DebugConfig, ProxyAppState, ProxyRetryConfig, ProxyTrafficState, ReadinessState, readyz};
+    use super::{
+        DebugConfig, ProxyAppState, ProxyRetryConfig, ProxyTrafficState, ReadinessState, readyz,
+    };
 
     pub(super) fn test_proxy_app_state() -> ProxyAppState {
         test_proxy_app_state_with_lb_config(LoadBalancerConfig::default())
@@ -286,6 +292,8 @@ mod test_support {
                     .expect("load balancer should initialize"),
             ),
             metrics,
+            routing_reservation_ttl_min: std::time::Duration::from_millis(1),
+            routing_reservation_ttl_max: std::time::Duration::from_millis(1000),
             retry: ProxyRetryConfig::default(),
             debug_config: DebugConfig::default(),
         }
@@ -308,7 +316,10 @@ mod test_support {
         let ready_token = CancellationToken::new();
         app.readiness = ReadinessState::warming_up(ready_token.clone());
 
-        assert_eq!(readyz(State(app.clone())).await, StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(
+            readyz(State(app.clone())).await,
+            StatusCode::SERVICE_UNAVAILABLE
+        );
 
         ready_token.cancel();
 

src/libraries/rust/stargate/crates/stargate/src/http_proxy/attempt.rs

@@ -43,6 +43,16 @@ use super::upstream::{
 const HEADER_CHOSEN_INFERENCE_SERVER_URL: &str = "x-inference-server-url";
 const HEADER_CHOSEN_CLUSTER_ID: &str = "x-stargate-cluster-id";
 
+fn clamp_routing_reservation_ttl(
+    rtt: std::time::Duration,
+    min: std::time::Duration,
+    max: std::time::Duration,
+) -> std::time::Duration {
+    // Startup rejects min > max; `Duration::clamp` would panic on every
+    // dispatch if a programmatic config bypassed that check, so max wins.
+    rtt.max(min).min(max)
+}
+
 #[derive(Default)]
 pub(super) struct ProxyAttemptCounters {
     pub(super) attempt: u32,
@@ -112,11 +122,24 @@ impl ProxyRequestRun<'_> {
             }
         }
 
-        let reservation: Option<RoutingReservation> = selected.cluster.reserve_backend(
-            &chosen.registration,
-            self.request.request_inputs.input_tokens,
-            self.request.request_inputs.priority,
-        );
+        let reservation: Option<RoutingReservation> =
+            selected.cluster.reserve_backend_with_rtt_ttl(
+                &chosen.registration,
+                self.request.request_inputs.input_tokens,
+                self.request.request_inputs.priority,
+                |rtt| {
+                    clamp_routing_reservation_ttl(
+                        rtt,
+                        self.app.routing_reservation_ttl_min,
+                        self.app.routing_reservation_ttl_max,
+                    )
+                },
+                self.app.metrics.clone(),
+                &crate::routing_state::RoutingTargetKey::new(
+                    self.routing_key().map(str::to_owned),
+                    self.model_id(),
+                ),
+            );
         record_proxy_attempt_start(self, selected, chosen);
 
         let upstream_span = proxy_upstream_attempt_span(self, selected, chosen);
@@ -532,6 +555,7 @@ fn upstream_status(result: &Result<UpstreamStreamingResponse, StatusCode>) -> St
 mod tests {
     use super::*;
     use axum::http::{HeaderMap, Method};
+    use prometheus::Encoder;
     use stargate_proto::pb::{InferenceServerStatus, ModelStats};
     use stargate_protocol::tunnel_contract::HEADER_STARGATE_RETRY_AFTER_MS;
     use std::time::Duration;
@@ -541,10 +565,237 @@ mod tests {
     use super::super::retry::ReplayableRequestBody;
     use super::super::run::PreparedProxyRequest;
     use super::super::test_support::test_proxy_app_state;
+    use crate::load_balancer::LoadBalancerCandidateChoice;
     use crate::routing_state::{
-        RegistrationIdentity, RoutingTargetKey, test_registration_generation,
+        RegistrationIdentity, RoutingTargetKey, StargateState, test_registration_generation,
     };
 
+    #[test]
+    fn reservation_ttl_is_clamped_to_the_configured_bounds() {
+        let min = Duration::from_millis(1);
+        let max = Duration::from_millis(1000);
+
+        assert_eq!(clamp_routing_reservation_ttl(Duration::ZERO, min, max), min);
+        assert_eq!(
+            clamp_routing_reservation_ttl(Duration::from_millis(60), min, max),
+            Duration::from_millis(60)
+        );
+        assert_eq!(
+            clamp_routing_reservation_ttl(Duration::from_secs(5), min, max),
+            max
+        );
+    }
+
+    #[test]
+    fn reservation_ttl_clamp_does_not_panic_on_inverted_bounds() {
+        let min = Duration::from_millis(20);
+        let max = Duration::from_millis(10);
+
+        for rtt in [Duration::ZERO, Duration::from_millis(15), Duration::MAX] {
+            assert_eq!(clamp_routing_reservation_ttl(rtt, min, max), max);
+        }
+    }
+
+    #[tokio::test]
+    async fn proxy_reservation_ttl_uses_the_selected_backends_rtt() {
+        let mut app = test_proxy_app_state();
+        app.state = Arc::new(StargateState::new_with_metrics(app.metrics.clone()));
+        let target = RoutingTargetKey::new(Some("tenant-a".to_string()), "model-a");
+
+        for (backend_id, port, rtt_ms) in [("fast", 5001, 10), ("slow", 5002, 100)] {
+            let identity = RegistrationIdentity {
+                inference_server_id: backend_id.to_string(),
+                cluster_id: "cluster-a".to_string(),
+                inference_server_url: format!("quic://127.0.0.1:{port}"),
+                routing_key: target.routing_key.clone(),
+                reverse_tunnel: true,
+            };
+            let registration = app.state.begin_registration(&identity).unwrap();
+            let update = stargate_proto::pb::InferenceServerRegistration {
+                inference_server_id: identity.inference_server_id.clone(),
+                cluster_id: identity.cluster_id.clone(),
+                inference_server_url: identity.inference_server_url.clone(),
+                models: std::collections::HashMap::from([(
+                    target.model_id.clone(),
+                    stargate_proto::pb::InferenceServerModelRegistration {
+                        stats: Some(ModelStats {
+                            last_mean_input_tps: 100.0,
+                            queue_time_estimate_ms_by_priority: std::collections::HashMap::from([
+                                (0, 10),
+                            ]),
+                            ..ModelStats::default()
+                        }),
+                        status: InferenceServerStatus::Active.into(),
+                    },
+                )]),
+                reverse_tunnel: true,
+            };
+            app.state
+                .apply_registration_update(
+                    &registration,
+                    &update,
+                    true,
+                    Some(Duration::from_millis(rtt_ms)),
+                )
+                .await;
+        }
+
+        let candidates = app.state.candidates_for_target(&target).await;
+        let chosen = Arc::new(
+            candidates
+                .into_iter()
+                .min_by_key(|candidate| candidate.rtt)
+                .expect("both backends should be routable"),
+        );
+        assert_eq!(chosen.inference_server_id, "fast");
+        assert_eq!(chosen.rtt, Duration::from_millis(10));
+
+        let target_snapshot = app
+            .state
+            .routing_target_snapshot(&target)
+            .await
+            .expect("registered backends should produce a target snapshot");
+        let resolution = app
+            .lb_router
+            .resolve_algorithm_override(&target.model_id, None)
+            .unwrap();
+        let selected = super::super::run::SelectedClusterRun::new(
+            target_snapshot,
+            resolution.selection(LoadBalancerCandidateChoice::with_rank_depth_1(0)),
+            0,
+        );
+        let request = PreparedProxyRequest {
+            lb_resolution: resolution,
+            request_inputs: ProxyRequestInputs {
+                target,
+                input_tokens: 10,
+                priority: 0,
+                max_wait_ms: None,
+                request_slo_ms: None,
+                cache_affinity_key: None,
+                routing_algorithm_override: None,
+            },
+            endpoint_name: "chat_completions",
+            method: Method::POST,
+            path_and_query: "/v1/chat/completions".to_string(),
+            forwarded_headers: HeaderMap::new(),
+            retry_deadline: None,
+            request_start: Instant::now(),
+            replay_body: ReplayableRequestBody::new(&HeaderMap::new(), Body::empty(), 1024)
+                .unwrap(),
+        };
+        let mut run = ProxyRequestRun::new(&app, request);
+        let _ = run.run_proxy_attempt(&selected, &chosen).await;
+
+        let mut encoded = Vec::new();
+        prometheus::TextEncoder::new()
+            .encode(&app.metrics.registry().gather(), &mut encoded)
+            .unwrap();
+        let metrics = String::from_utf8(encoded).unwrap();
+        assert!(metrics.contains(
+            "stargate_routing_reservation_ttl_seconds_sum{model=\"model-a\",routing_key=\"tenant-a\"} 0.01"
+        ));
+        assert!(metrics.contains(
+            "stargate_routing_reservations_active{model=\"model-a\",routing_key=\"tenant-a\"} 1"
+        ));
+    }
+
+    #[tokio::test]
+    async fn same_backend_retry_reserves_with_the_current_backend_rtt() {
+        let mut app = test_proxy_app_state();
+        app.state = Arc::new(StargateState::new_with_metrics(app.metrics.clone()));
+        let target = RoutingTargetKey::new(Some("tenant-a".to_string()), "model-a");
+        let identity = RegistrationIdentity {
+            inference_server_id: "backend-a".to_string(),
+            cluster_id: "cluster-a".to_string(),
+            inference_server_url: "quic://127.0.0.1:5001".to_string(),
+            routing_key: target.routing_key.clone(),
+            reverse_tunnel: true,
+        };
+        let registration = app.state.begin_registration(&identity).unwrap();
+        let update = stargate_proto::pb::InferenceServerRegistration {
+            inference_server_id: identity.inference_server_id.clone(),
+            cluster_id: identity.cluster_id.clone(),
+            inference_server_url: identity.inference_server_url.clone(),
+            models: std::collections::HashMap::from([(
+                target.model_id.clone(),
+                stargate_proto::pb::InferenceServerModelRegistration {
+                    stats: Some(ModelStats::default()),
+                    status: InferenceServerStatus::Active.into(),
+                },
+            )]),
+            reverse_tunnel: true,
+        };
+        app.state
+            .apply_registration_update(
+                &registration,
+                &update,
+                true,
+                Some(Duration::from_millis(60)),
+            )
+            .await;
+        let resolution = app
+            .lb_router
+            .resolve_algorithm_override(&target.model_id, None)
+            .unwrap();
+        let selected = super::super::run::SelectedClusterRun::new(
+            app.state
+                .routing_target_snapshot(&target)
+                .await
+                .expect("registered backend should produce a target snapshot"),
+            resolution.selection(LoadBalancerCandidateChoice::with_rank_depth_1(0)),
+            0,
+        );
+        // A same-backend retry reuses the snapshot chosen for the first attempt.
+        let retained = selected
+            .cluster
+            .select_backend(&std::collections::HashSet::new())
+            .expect("registered backend should be selectable");
+        assert_eq!(retained.rtt, Duration::from_millis(60));
+
+        app.state
+            .apply_registration_update(
+                &registration,
+                &update,
+                true,
+                Some(Duration::from_millis(200)),
+            )
+            .await;
+        let request = PreparedProxyRequest {
+            lb_resolution: resolution,
+            request_inputs: ProxyRequestInputs {
+                target,
+                input_tokens: 10,
+                priority: 0,
+                max_wait_ms: None,
+                request_slo_ms: None,
+                cache_affinity_key: None,
+                routing_algorithm_override: None,
+            },
+            endpoint_name: "chat_completions",
+            method: Method::POST,
+            path_and_query: "/v1/chat/completions".to_string(),
+            forwarded_headers: HeaderMap::new(),
+            retry_deadline: None,
+            request_start: Instant::now(),
+            replay_body: ReplayableRequestBody::new(&HeaderMap::new(), Body::empty(), 1024)
+                .unwrap(),
+        };
+        let mut run = ProxyRequestRun::new(&app, request);
+        let _ = run.run_proxy_attempt(&selected, &retained).await;
+
+        let ttl = app
+            .metrics
+            .routing_reservation_ttl_seconds(Some("tenant-a"), "model-a");
+        assert_eq!(ttl.get_sample_count(), 1);
+        assert_eq!(
+            Duration::from_secs_f64(ttl.get_sample_sum()),
+            Duration::from_millis(200),
+            "a new dispatch must use the backend's current RTT, not the retained snapshot"
+        );
+        app.state.end_registration(registration).await;
+    }
+
     fn finish_response(
         disposition: FinalRetryDisposition,
         status: StatusCode,