[Bugfix]: strip @rank suffix in DP intra-node GET proxy URLs

src/routers/http/router.rs

@@ -428,13 +428,20 @@ impl Router {
 
         match self.select_first_worker() {
             Ok(worker_url) => {
-                let url = format!("{}/{}", worker_url, endpoint);
+                let (base_url, dp_rank) = dp_utils::parse_worker_url(&worker_url);
+                let url = format!("{}/{}", base_url, endpoint);
                 let route_name = format!("/{}", endpoint);
-                let mut request_builder = self.client.get(&url);
+                let mut request_builder =
+                    dp_utils::add_dp_rank_header(self.client.get(&url), dp_rank);
+
                 for (name, value) in headers {
                     let name_lc = name.to_lowercase();
+                    // When the router selects a DP rank, it owns the
+                    // X-data-parallel-rank header: skip any client-supplied
+                    // value so the worker sees exactly one rank (ours).
                     if name_lc != "content-type"
                         && name_lc != "content-length"
+                        && !(dp_rank.is_some() && name_lc == "x-data-parallel-rank")
                         && !header_utils::TRACE_HEADER_NAMES.contains(&name_lc.as_str())
                     {
                         request_builder = request_builder.header(name, value);

tests/common/mock_worker.rs

@@ -684,9 +684,15 @@ async fn flush_cache_handler(State(config): State<Arc<RwLock<MockWorkerConfig>>>
     .into_response()
 }
 
-async fn v1_models_handler(State(config): State<Arc<RwLock<MockWorkerConfig>>>) -> Response {
+async fn v1_models_handler(
+    State(config): State<Arc<RwLock<MockWorkerConfig>>>,
+    headers: axum::http::HeaderMap,
+) -> Response {
     let config = config.read().await;
 
+    // Capture request for test inspection (e.g. X-data-parallel-rank)
+    capture_request(config.port, "/v1/models", &headers);
+
     if should_fail(&config).await {
         return (
             StatusCode::INTERNAL_SERVER_ERROR,
@@ -872,10 +878,14 @@ impl Default for MockWorkerConfig {
 // --- Request header capture for verifying router behavior (e.g., X-data-parallel-rank) ---
 
 /// A captured request with headers and path
+///
+/// `headers` maps each header name to *all* values received for that name,
+/// so duplicate headers (e.g. a client-supplied and a router-injected
+/// X-data-parallel-rank) are preserved for inspection.
 #[derive(Debug, Clone)]
 pub struct CapturedRequest {
     pub path: String,
-    pub headers: HashMap<String, String>,
+    pub headers: HashMap<String, Vec<String>>,
 }
 
 static REQ_CAPTURE_STORE: OnceLock<Mutex<HashMap<u16, Vec<CapturedRequest>>>> = OnceLock::new();
@@ -886,17 +896,18 @@ fn get_capture_store() -> &'static Mutex<HashMap<u16, Vec<CapturedRequest>>> {
 
 /// Record a request for a given worker port
 pub fn capture_request(port: u16, path: &str, headers: &axum::http::HeaderMap) {
+    let mut captured_headers: HashMap<String, Vec<String>> = HashMap::new();
+    for (name, value) in headers.iter() {
+        if let Ok(v) = value.to_str() {
+            captured_headers
+                .entry(name.as_str().to_string())
+                .or_default()
+                .push(v.to_string());
+        }
+    }
     let captured = CapturedRequest {
         path: path.to_string(),
-        headers: headers
-            .iter()
-            .filter_map(|(name, value)| {
-                value
-                    .to_str()
-                    .ok()
-                    .map(|v| (name.as_str().to_string(), v.to_string()))
-            })
-            .collect(),
+        headers: captured_headers,
     };
     let mut store = get_capture_store().lock().unwrap();
     store.entry(port).or_default().push(captured);

tests/otel_integration_test.rs

@@ -207,6 +207,7 @@ async fn test_otel_integration() {
         let backend_tp = backend_req
             .headers
             .get("traceparent")
+            .and_then(|values| values.first())
             .expect("Backend should receive traceparent header");
         let parts: Vec<&str> = backend_tp.split('-').collect();
         assert_eq!(parts.len(), 4, "traceparent should have 4 parts");
@@ -223,6 +224,7 @@ async fn test_otel_integration() {
         let backend_baggage = backend_req
             .headers
             .get("baggage")
+            .and_then(|values| values.first())
             .expect("Backend should receive baggage header");
         assert!(
             backend_baggage.contains("userId=alice"),

tests/test_dp_routing.rs

@@ -449,6 +449,142 @@ mod dp_e2e_tests {
         worker.stop().await;
     }
 
+    #[tokio::test]
+    async fn test_regular_router_dp2_get_v1_models() {
+        let mut worker = MockWorker::new(MockWorkerConfig::default());
+        let worker_url = worker.start().await.unwrap();
+        let port: u16 = worker_url.split(':').next_back().unwrap().parse().unwrap();
+        clear_captured_requests(port);
+
+        let config = make_regular_config(vec![worker_url.clone()], 2);
+        let app_context = common::create_test_context(config.clone());
+        let router = RouterFactory::create_router(&app_context).await.unwrap();
+        let router = Arc::from(router);
+
+        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
+
+        let app = common::test_app::create_test_app(Arc::clone(&router), Client::new(), &config);
+
+        let req = Request::builder()
+            .uri("/v1/models")
+            .body(Body::empty())
+            .unwrap();
+
+        let resp = app.oneshot(req).await.unwrap();
+        assert_eq!(
+            resp.status().as_u16(),
+            200,
+            "GET /v1/models must succeed when DP > 1 (got {}). @rank in the worker URL was misinterpreted as userinfo.",
+            resp.status()
+        );
+
+        // The 200 alone could be misleading if a fallback were added later;
+        // assert the body is the mock worker's model list.
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(
+            body["data"][0]["id"].as_str(),
+            Some("mock-model"),
+            "GET /v1/models should return the worker's model list. Body: {:?}",
+            body
+        );
+
+        // Verify the forwarded X-data-parallel-rank matches the selected
+        // worker's rank (0 or 1 here; both DP ranks share this mock worker,
+        // and worker selection order is not deterministic).
+        let captured = get_captured_requests(port);
+        let models_req = captured
+            .iter()
+            .find(|r| r.path == "/v1/models")
+            .expect("Mock worker should have received the GET /v1/models request");
+        let rank_values = models_req
+            .headers
+            .get("x-data-parallel-rank")
+            .expect("GET proxy must forward X-data-parallel-rank when DP > 1");
+        assert_eq!(
+            rank_values.len(),
+            1,
+            "exactly one X-data-parallel-rank value expected, got {:?}",
+            rank_values
+        );
+        let rank: usize = rank_values[0]
+            .parse()
+            .expect("X-data-parallel-rank must be numeric");
+        assert!(
+            rank < 2,
+            "forwarded rank must be one of the router's DP ranks (0 or 1), got {}",
+            rank
+        );
+
+        worker.stop().await;
+    }
+
+    #[tokio::test]
+    async fn test_regular_router_dp2_get_v1_models_overrides_client_dp_rank() {
+        // A client-supplied X-data-parallel-rank must not leak through or
+        // duplicate the router-selected rank: the worker should receive
+        // exactly one value, matching the router's selection.
+        let mut worker = MockWorker::new(MockWorkerConfig::default());
+        let worker_url = worker.start().await.unwrap();
+        let port: u16 = worker_url.split(':').next_back().unwrap().parse().unwrap();
+        clear_captured_requests(port);
+
+        let config = make_regular_config(vec![worker_url.clone()], 2);
+        let app_context = common::create_test_context(config.clone());
+        let router = RouterFactory::create_router(&app_context).await.unwrap();
+        let router = Arc::from(router);
+
+        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
+
+        let app = common::test_app::create_test_app(Arc::clone(&router), Client::new(), &config);
+
+        let req = Request::builder()
+            .uri("/v1/models")
+            .header("x-data-parallel-rank", "99")
+            .body(Body::empty())
+            .unwrap();
+
+        let resp = app.oneshot(req).await.unwrap();
+        assert_eq!(
+            resp.status().as_u16(),
+            200,
+            "GET /v1/models must succeed even when the client sends a conflicting rank (got {})",
+            resp.status()
+        );
+
+        let captured = get_captured_requests(port);
+        let models_req = captured
+            .iter()
+            .find(|r| r.path == "/v1/models")
+            .expect("Mock worker should have received the GET /v1/models request");
+        let rank_values = models_req
+            .headers
+            .get("x-data-parallel-rank")
+            .expect("router-selected X-data-parallel-rank must be forwarded");
+        assert_eq!(
+            rank_values.len(),
+            1,
+            "client-supplied rank must be filtered out; worker received {:?}",
+            rank_values
+        );
+        assert_ne!(
+            rank_values[0], "99",
+            "worker must receive the router's selection, not the client's value"
+        );
+        let rank: usize = rank_values[0]
+            .parse()
+            .expect("X-data-parallel-rank must be numeric");
+        assert!(
+            rank < 2,
+            "forwarded rank must be one of the router's DP ranks (0 or 1), got {}",
+            rank
+        );
+
+        worker.stop().await;
+    }
+
     // -----------------------------------------------------------------
     // Regular Router + DP > 1: worker registry verification
     // -----------------------------------------------------------------