From 95a8b364e24468f05f9e983684f491113a7638ad Mon Sep 17 00:00:00 2001 From: along Date: Mon, 3 Aug 2026 21:24:53 -0700 Subject: [PATCH 01/17] feat(pylon): derive dynamo request priority from x-priority Pylon owns the engine-facing Dynamo header contract. On every tunneled inference request it now strips inbound x-dynamo-request-* headers so client-supplied values never reach the engine, and derives x-dynamo-request-priority from x-priority when that header is present (i32::MAX - min(x, i32::MAX); absent stays absent). Derivation sits behind the default-on --pylon-derive-dynamo-priority flag as a kill switch; the strip is unconditional. The emitted value is logged with the request id and recorded on the upstream request span. mock-dynamo records the priority headers seen on the latest request per endpoint and model in its /test-control snapshot so cluster QA can assert what actually reached the engine. The pylon and mock-dynamo crates gain rust_test Bazel targets; their in-crate tests previously did not run in CI. Refs: NVIDIA/nvcf#620 Signed-off-by: along --- .../stargate/crates/mock-dynamo/BUILD.bazel | 8 +- .../stargate/crates/mock-dynamo/src/openai.rs | 11 +- .../crates/mock-dynamo/src/test_control.rs | 59 +++++++ .../stargate/crates/mock-dynamo/src/tests.rs | 66 +++++++- .../pylon-lib/src/quic_http_tunnel/core.rs | 65 +++++++- .../src/quic_http_tunnel/raw_quic.rs | 1 + .../pylon-lib/src/quic_http_tunnel/tests.rs | 148 +++++++++++++++++- .../rust/stargate/crates/pylon/BUILD.bazel | 8 +- .../rust/stargate/crates/pylon/src/main.rs | 27 +++- .../rust/stargate/crates/pylon/src/startup.rs | 12 ++ 10 files changed, 389 insertions(+), 16 deletions(-) diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel b/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel index 05a653501..68ad1f0b3 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel +++ b/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel @@ -6,7 +6,7 @@ # helper only). load("@stargate_crates//:defs.bzl", "aliases", "all_crate_deps") -load("@rules_rust//rust:defs.bzl", "rust_binary") +load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_test") rust_binary( name = "mock-dynamo", @@ -17,3 +17,9 @@ rust_binary( visibility = ["//visibility:public"], deps = all_crate_deps(normal = True), ) + +rust_test( + name = "mock-dynamo_test", + crate = ":mock-dynamo", + deps = all_crate_deps(normal_dev = True), +) diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs index 7604e2307..df718ec12 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs @@ -26,7 +26,9 @@ use tracing::info; use crate::AppState; use crate::kv_cache::{KvCacheAccess, KvCacheStats, insert_kv_cache_headers}; use crate::stats_stream::StatsStreamEvent; -use crate::test_control::{TestEndpoint, TestRequestClass, request_class}; +use crate::test_control::{ + TestEndpoint, TestRequestClass, recorded_priority_headers, request_class, +}; use crate::timing::{ embedding_item_count, jitter_ms, non_streaming_delay, optional_header, prefill_delay, request_embedding_tokens, request_input_tokens, request_output_tokens, response_input_tokens, @@ -420,7 +422,12 @@ impl AppState { async fn record_request(&self, headers: &HeaderMap, endpoint: TestEndpoint, model: &str) { let request_class = request_class(headers); self.test_control - .record_request(endpoint, model, request_class) + .record_request( + endpoint, + model, + request_class, + recorded_priority_headers(headers), + ) .await; if request_class == TestRequestClass::PylonGenerated { self.test_control.wait_for_bringup_release(model).await; diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs index f77a23bfb..ad373920c 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs @@ -56,12 +56,22 @@ pub(crate) struct ModelTestControlUpdate { type TestCounterKey = (TestEndpoint, String, TestRequestClass); +/// Raw priority header values seen on the most recent request, kept so tunnel +/// and gateway tests can assert what actually reached the mock engine. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +pub(crate) struct RecordedPriorityHeaders { + pub(crate) x_priority: Option, + pub(crate) dynamo_request_priority: Option, + pub(crate) dynamo_request_strict_priority: Option, +} + #[derive(Debug, Default)] struct TestControlInner { discovered_models: BTreeSet, model_discovery_requests: u64, models: BTreeMap, counters: BTreeMap, + priority_headers: BTreeMap<(TestEndpoint, String), RecordedPriorityHeaders>, } #[derive(Debug, Clone, Default)] @@ -78,12 +88,20 @@ pub(crate) struct TestCounterSnapshot { pub(crate) count: u64, } +#[derive(Debug, Clone, Serialize)] +pub(crate) struct PriorityHeadersSnapshot { + pub(crate) endpoint: TestEndpoint, + pub(crate) model: String, + pub(crate) headers: RecordedPriorityHeaders, +} + #[derive(Debug, Clone, Serialize)] pub(crate) struct TestControlSnapshot { pub(crate) discovered_models: Vec, pub(crate) model_discovery_requests: u64, pub(crate) models: BTreeMap, pub(crate) counters: Vec, + pub(crate) priority_headers: Vec, } impl TestControlState { @@ -170,6 +188,7 @@ impl TestControlState { endpoint: TestEndpoint, model: &str, request_class: TestRequestClass, + priority_headers: RecordedPriorityHeaders, ) { let mut inner = self.inner.lock().await; let count = inner @@ -177,6 +196,11 @@ impl TestControlState { .entry((endpoint, model.to_string(), request_class)) .or_default(); *count = count.saturating_add(1); + // Last-seen semantics: a request without priority headers overwrites + // earlier values, so the snapshot always reflects the latest request. + inner + .priority_headers + .insert((endpoint, model.to_string()), priority_headers); } pub(crate) async fn snapshot(&self) -> TestControlSnapshot { @@ -197,6 +221,15 @@ impl TestControlState { }, ) .collect(), + priority_headers: inner + .priority_headers + .iter() + .map(|((endpoint, model), headers)| PriorityHeadersSnapshot { + endpoint: *endpoint, + model: model.clone(), + headers: headers.clone(), + }) + .collect(), } } } @@ -235,6 +268,18 @@ impl TestControlSnapshot { }) .map_or(0, |counter| counter.count) } + + #[cfg(test)] + pub(crate) fn priority_headers( + &self, + endpoint: TestEndpoint, + model: &str, + ) -> Option<&RecordedPriorityHeaders> { + self.priority_headers + .iter() + .find(|entry| entry.endpoint == endpoint && entry.model == model) + .map(|entry| &entry.headers) + } } pub(crate) async fn update_model_test_control( @@ -252,6 +297,20 @@ pub(crate) async fn test_control_snapshot( Json(state.test_control.snapshot().await) } +pub(crate) fn recorded_priority_headers(headers: &HeaderMap) -> RecordedPriorityHeaders { + let header_value = |name: &str| { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(str::to_string) + }; + RecordedPriorityHeaders { + x_priority: header_value("x-priority"), + dynamo_request_priority: header_value("x-dynamo-request-priority"), + dynamo_request_strict_priority: header_value("x-dynamo-request-strict-priority"), + } +} + pub(crate) fn request_class(headers: &HeaderMap) -> TestRequestClass { match headers .get("x-request-id") diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs index d4ba751a0..786e3c5c9 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs @@ -191,7 +191,12 @@ async fn test_controls_count_endpoint_model_and_request_class() { ]; for (endpoint, model, request_class) in recorded { controls - .record_request(endpoint, model, request_class) + .record_request( + endpoint, + model, + request_class, + RecordedPriorityHeaders::default(), + ) .await; } @@ -308,6 +313,65 @@ async fn test_control_http_api_updates_one_model_and_reports_request_counters() server.abort(); } +#[tokio::test] +async fn test_control_snapshot_reports_last_seen_priority_headers() { + let state = test_state(); + let observed_control = state.test_control.clone(); + let app = Router::new() + .route("/v1/chat/completions", post(chat_completions)) + .route("/test-control", get(test_control_snapshot)) + .with_state(state); + let (addr, server) = spawn_test_app(app).await; + + let body = r#"{"model":"model-a","messages":[],"max_tokens":1,"stream":false}"#; + let prioritized_response = json_response( + addr, + "POST", + "/v1/chat/completions", + "connection: close\r\nx-request-id: user-8\r\nx-priority: 7\r\nx-dynamo-request-priority: 2147483640", + body, + ) + .await; + assert!(prioritized_response.starts_with("HTTP/1.1 200 OK")); + + let snapshot = observed_control.snapshot().await; + assert_eq!( + snapshot.priority_headers(TestEndpoint::ChatCompletions, "model-a"), + Some(&RecordedPriorityHeaders { + x_priority: Some("7".to_string()), + dynamo_request_priority: Some("2147483640".to_string()), + dynamo_request_strict_priority: None, + }) + ); + + let snapshot_response = raw_http_request( + addr, + &format!("GET /test-control HTTP/1.1\r\nhost: {addr}\r\nconnection: close\r\n\r\n"), + ) + .await; + assert!(snapshot_response.contains(r#""dynamo_request_priority":"2147483640""#)); + + // A later request without priority headers overwrites the recorded values, + // so the snapshot always reflects the latest request. + let plain_response = json_response( + addr, + "POST", + "/v1/chat/completions", + "connection: close\r\nx-request-id: user-9", + body, + ) + .await; + assert!(plain_response.starts_with("HTTP/1.1 200 OK")); + + let snapshot = observed_control.snapshot().await; + assert_eq!( + snapshot.priority_headers(TestEndpoint::ChatCompletions, "model-a"), + Some(&RecordedPriorityHeaders::default()) + ); + + server.abort(); +} + #[tokio::test] async fn model_discovery_http_api_returns_and_replaces_authoritative_models() { let state = test_state(); diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index f5b89f5d6..cf362ca00 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -27,8 +27,9 @@ use reqwest::{Client, Error as ReqwestError, Method, Response, StatusCode}; use sonic_rs::JsonValueTrait; use stargate_protocol::common::is_hop_by_hop_header; use stargate_protocol::tunnel_contract::{ - HEADER_MODEL, HEADER_STARGATE_EXPECTED_QUEUE_MS, HEADER_STARGATE_RETRY_AFTER_MS, - HEADER_STARGATE_RETRY_REASON, HEADER_STARGATE_RETRYABLE, HEADER_STARGATE_UPSTREAM_RETRYABLE, + HEADER_MODEL, HEADER_PRIORITY, HEADER_REQUEST_ID, HEADER_STARGATE_EXPECTED_QUEUE_MS, + HEADER_STARGATE_RETRY_AFTER_MS, HEADER_STARGATE_RETRY_REASON, HEADER_STARGATE_RETRYABLE, + HEADER_STARGATE_UPSTREAM_RETRYABLE, }; use stargate_telemetry::{ inject_trace_context, parent_context_from_headers, traceparent_from_headers, @@ -105,6 +106,9 @@ pub struct TunnelForwardingConfig { pub request_quality_monitor: RequestQualityMonitorConfig, pub retry: PylonRetryConfig, pub queue_mismatch_retry: PylonQueueMismatchRetryConfig, + /// Derive x-dynamo-request-priority for the upstream engine from x-priority. + /// Inbound x-dynamo-request-* headers are stripped regardless of this gate. + pub derive_dynamo_priority: bool, pub metrics: Option>, #[cfg(test)] pub webtransport_stream_header_wait_tx: Option>, @@ -121,6 +125,7 @@ impl Default for TunnelForwardingConfig { request_quality_monitor: RequestQualityMonitorConfig::default(), retry: PylonRetryConfig::default(), queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(), + derive_dynamo_priority: true, metrics: None, #[cfg(test)] webtransport_stream_header_wait_tx: None, @@ -141,6 +146,7 @@ pub(super) struct TunnelServerApp { pub(super) request_quality_monitor: RequestQualityMonitorConfig, pub(super) retry: PylonRetryConfig, pub(super) queue_mismatch_retry: PylonQueueMismatchRetryConfig, + pub(super) derive_dynamo_priority: bool, pub(super) metrics: Option>, #[cfg(test)] pub(super) webtransport_stream_header_wait_tx: Option>, @@ -164,6 +170,7 @@ impl TunnelServerApp { request_quality_monitor: forwarding.request_quality_monitor, retry: forwarding.retry, queue_mismatch_retry: forwarding.queue_mismatch_retry, + derive_dynamo_priority: forwarding.derive_dynamo_priority, metrics: forwarding.metrics, #[cfg(test)] webtransport_stream_header_wait_tx: forwarding.webtransport_stream_header_wait_tx, @@ -710,6 +717,7 @@ async fn send_upstream_request( inference_server.id = %app.inference_server_id, upstream.status = field::Empty, upstream.error = field::Empty, + dynamo.request_priority = field::Empty, ); let _ = span.set_parent(pylon_upstream_parent_context(request_headers)); if let Some(otel_parent) = otel_parent_from_headers(request_headers) { @@ -725,6 +733,27 @@ async fn send_upstream_request( upstream_headers.append(name, value.clone()); } } + // `traced` excludes health requests, which skip header validation and are + // not client inference traffic; nothing to derive for them. + if app.derive_dynamo_priority && traced { + if let Some(priority) = x_priority_header_value(request_headers) { + let dynamo_priority = dynamo_request_priority(priority); + upstream_headers.insert( + HeaderName::from_static(HEADER_DYNAMO_REQUEST_PRIORITY), + HeaderValue::from(dynamo_priority), + ); + span.record("dynamo.request_priority", dynamo_priority); + tracing::info!( + request_id = request_headers + .get(HEADER_REQUEST_ID) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(), + priority, + dynamo.request_priority = dynamo_priority, + "derived dynamo request priority" + ); + } + } if traced { inject_trace_context(&mut upstream_headers, &span.context()); } @@ -1058,12 +1087,44 @@ pub(super) fn join_base_path(base: &str, path_and_query: &str) -> Result bool { !is_tunnel_control_header(name, retry) + && !is_dynamo_request_header(name) && !matches!( name.as_str(), "host" | "x-method" | "x-path" | HEADER_STARGATE_EXPECTED_QUEUE_MS ) } +/// Engine-facing priority header in the Dynamo contract; Pylon owns this +/// contract, so the constant stays out of the shared tunnel contract. +pub(super) const HEADER_DYNAMO_REQUEST_PRIORITY: &str = "x-dynamo-request-priority"; +const DYNAMO_REQUEST_HEADER_PREFIX: &str = "x-dynamo-request-"; + +/// Client-supplied Dynamo request headers never reach the engine; Pylon is +/// the only writer of x-dynamo-request-* values. +pub(super) fn is_dynamo_request_header(name: &HeaderName) -> bool { + name.as_str().starts_with(DYNAMO_REQUEST_HEADER_PREFIX) +} + +/// Dynamo schedules on an i32 where higher wins and silently drops values +/// that do not parse as i32, while x-priority is a u32 where lower wins, so +/// the mapping inverts and clamps to keep every configured priority valid. +pub(super) fn dynamo_request_priority(priority: u32) -> i32 { + i32::MAX - i32::try_from(priority).unwrap_or(i32::MAX) +} + +/// Absent stays absent: a request without x-priority must not be promoted to +/// maximum engine priority, so this reads the raw header instead of the +/// parsed default of 0. Malformed values were already rejected upstream. +pub(super) fn x_priority_header_value(headers: &HeaderMap) -> Option { + headers + .get(HEADER_PRIORITY)? + .to_str() + .ok()? + .trim() + .parse() + .ok() +} + pub(super) fn should_forward_response_header(name: &HeaderName, retry: &PylonRetryConfig) -> bool { !is_tunnel_control_header(name, retry) && name != CONTENT_LENGTH } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs index df26999b8..725518bfd 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs @@ -171,6 +171,7 @@ mod tests { request_quality_monitor: RequestQualityMonitorConfig::default(), retry: PylonRetryConfig::default(), queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(), + derive_dynamo_priority: true, metrics: None, #[cfg(test)] webtransport_stream_header_wait_tx: None, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index 6bdf49021..c06f284fe 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -14,10 +14,10 @@ // limitations under the License. use super::core::{ - MAX_SPECULATIVE_REQUEST_BODY_PREALLOC_BYTES, TunnelServerApp, extend_body_from_buf, - is_health_request_path, otel_parent_from_headers, pylon_upstream_parent_context, - request_body_buffer, request_body_capacity, should_forward_header, - should_forward_response_header, + MAX_SPECULATIVE_REQUEST_BODY_PREALLOC_BYTES, TunnelServerApp, dynamo_request_priority, + extend_body_from_buf, is_health_request_path, otel_parent_from_headers, + pylon_upstream_parent_context, request_body_buffer, request_body_capacity, + should_forward_header, should_forward_response_header, x_priority_header_value, }; use super::endpoint::{ build_trusted_client_config, derive_sni, make_server_config, target_authority, @@ -477,6 +477,9 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() "X-Method", "X-Path", "X-Stargate-Expected-Queue-Ms", + "X-Dynamo-Request-Priority", + "X-Dynamo-Request-Strict-Priority", + "X-Dynamo-Request-Anything", ] .into_iter() .chain(RETRY_CONTROL_REQUEST_HEADERS) @@ -486,13 +489,39 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() &retry )); } - assert!(should_forward_header( - &HeaderName::from_bytes(b"X-Request-Id")?, - &retry - )); + for name in [b"X-Request-Id".as_slice(), b"X-Priority", b"X-Dynamo-Nvext"] { + assert!(should_forward_header( + &HeaderName::from_bytes(name)?, + &retry + )); + } Ok(()) } +#[test] +fn pylon_dynamo_request_priority_inverts_and_clamps_to_i32() { + assert_eq!(dynamo_request_priority(0), i32::MAX); + assert_eq!(dynamo_request_priority(7), i32::MAX - 7); + assert_eq!(dynamo_request_priority(i32::MAX as u32), 0); + assert_eq!(dynamo_request_priority(i32::MAX as u32 + 1), 0); + assert_eq!(dynamo_request_priority(u32::MAX), 0); +} + +#[test] +fn pylon_x_priority_header_value_distinguishes_absent_from_zero() { + let mut headers = HeaderMap::new(); + assert_eq!(x_priority_header_value(&headers), None); + + headers.insert("x-priority", "0".parse().unwrap()); + assert_eq!(x_priority_header_value(&headers), Some(0)); + + headers.insert("x-priority", " 7 ".parse().unwrap()); + assert_eq!(x_priority_header_value(&headers), Some(7)); + + headers.insert("x-priority", "not-a-priority".parse().unwrap()); + assert_eq!(x_priority_header_value(&headers), None); +} + #[test] fn pylon_trace_context_extracts_remote_parent() -> Result<()> { opentelemetry::global::set_text_map_propagator( @@ -1633,6 +1662,109 @@ async fn quic_tunnel_forwards_to_http_backend() { tunnel.shutdown().await; } +fn dynamo_priority_echo_router() -> Router { + Router::new().route( + "/v1/chat/completions", + post(|req: Request| async move { + let dynamo_priority = req + .headers() + .get("x-dynamo-request-priority") + .and_then(|value| value.to_str().ok()) + .unwrap_or("absent") + .to_string(); + let saw_spoofed_dynamo_header = req + .headers() + .contains_key("x-dynamo-request-strict-priority"); + let mut sse = axum::response::Sse::new(async_stream::stream! { + yield Ok::<_, std::convert::Infallible>( + Event::default().data(r#"{"object":"chat.completion.chunk","choices":[{"delta":{"content":"ok"}}]}"#) + ); + yield Ok::<_, std::convert::Infallible>(Event::default().data("[DONE]")); + }) + .into_response(); + sse.headers_mut().insert( + HeaderName::from_static("x-echo-dynamo-priority"), + HeaderValue::from_str(&dynamo_priority).unwrap(), + ); + sse.headers_mut().insert( + HeaderName::from_static("x-saw-spoofed-dynamo-header"), + HeaderValue::from_str(&saw_spoofed_dynamo_header.to_string()).unwrap(), + ); + *sse.status_mut() = StatusCode::OK; + sse + }), + ) +} + +#[tokio::test] +async fn quic_tunnel_derives_dynamo_priority_from_x_priority() { + let (config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; + let mut tunnel = RawTunnelTest::start(config).await; + + let mut headers = + tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-1", "11"); + headers.insert("x-priority", "7".parse().unwrap()); + headers.insert("x-dynamo-request-priority", "42".parse().unwrap()); + headers.insert("x-dynamo-request-strict-priority", "true".parse().unwrap()); + tunnel + .send(headers, br#"{"messages":[],"stream":true}"#) + .await; + + let response_headers = tunnel.response_head(StatusCode::OK).await; + assert_eq!( + response_headers + .get("x-echo-dynamo-priority") + .unwrap() + .to_str() + .unwrap(), + (i32::MAX - 7).to_string() + ); + assert_eq!(response_headers["x-saw-spoofed-dynamo-header"], "false"); + + tunnel.shutdown().await; +} + +#[tokio::test] +async fn quic_tunnel_omits_dynamo_priority_without_x_priority() { + let (config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; + let mut tunnel = RawTunnelTest::start(config).await; + + let mut headers = + tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-2", "11"); + headers.insert("x-dynamo-request-priority", "42".parse().unwrap()); + tunnel + .send(headers, br#"{"messages":[],"stream":true}"#) + .await; + + let response_headers = tunnel.response_head(StatusCode::OK).await; + assert_eq!(response_headers["x-echo-dynamo-priority"], "absent"); + + tunnel.shutdown().await; +} + +#[tokio::test] +async fn quic_tunnel_dynamo_priority_derivation_can_be_disabled() { + let (mut config, _metrics) = + metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; + config.forwarding.derive_dynamo_priority = false; + let mut tunnel = RawTunnelTest::start(config).await; + + let mut headers = + tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-3", "11"); + headers.insert("x-priority", "7".parse().unwrap()); + headers.insert("x-dynamo-request-strict-priority", "true".parse().unwrap()); + tunnel + .send(headers, br#"{"messages":[],"stream":true}"#) + .await; + + let response_headers = tunnel.response_head(StatusCode::OK).await; + assert_eq!(response_headers["x-echo-dynamo-priority"], "absent"); + // Stripping inbound x-dynamo-request-* headers is not gated by the flag. + assert_eq!(response_headers["x-saw-spoofed-dynamo-header"], "false"); + + tunnel.shutdown().await; +} + #[tokio::test] async fn quic_tunnel_rejects_pending_generation_before_upstream() { let upstream_hits = Arc::new(AtomicUsize::new(0)); diff --git a/src/libraries/rust/stargate/crates/pylon/BUILD.bazel b/src/libraries/rust/stargate/crates/pylon/BUILD.bazel index cf18a04d6..e0238a026 100644 --- a/src/libraries/rust/stargate/crates/pylon/BUILD.bazel +++ b/src/libraries/rust/stargate/crates/pylon/BUILD.bazel @@ -3,7 +3,7 @@ load("@rules_shell//shell:sh_test.bzl", "sh_test") load("@stargate_crates//:defs.bzl", "aliases", "all_crate_deps") -load("@rules_rust//rust:defs.bzl", "rust_binary") +load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_test") load("//rules/oci:defs.bzl", "rust_oci_image") # `all_crate_deps()` only resolves third-party crate_universe deps; @@ -28,6 +28,12 @@ rust_binary( deps = _WORKSPACE_DEPS + all_crate_deps(normal = True), ) +rust_test( + name = "pylon_test", + crate = ":pylon", + deps = _WORKSPACE_DEPS + all_crate_deps(normal_dev = True), +) + # Multi-arch OCI image. distroless/cc base, binary at /usr/local/bin/pylon. rust_oci_image( name = "image", diff --git a/src/libraries/rust/stargate/crates/pylon/src/main.rs b/src/libraries/rust/stargate/crates/pylon/src/main.rs index 335632df8..94b5db966 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/main.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/main.rs @@ -204,6 +204,14 @@ struct Args { /// Optional retry-after hint in milliseconds for local queue-mismatch retries #[arg(long, env = "PYLON_QUEUE_MISMATCH_RETRY_AFTER_MS", value_name = "MS")] pylon_queue_mismatch_retry_after_ms: Option, + /// Derive x-dynamo-request-priority for the upstream engine from x-priority + #[arg( + long, + action = clap::ArgAction::Set, + default_value_t = true, + env = "PYLON_DERIVE_DYNAMO_PRIORITY" + )] + pylon_derive_dynamo_priority: bool, /// Collect post-stream output quality metrics (gibberish checks) #[arg(long, default_value_t = false)] collect_quality_metrics: bool, @@ -243,7 +251,7 @@ async fn main() -> Result<()> { mod tests { use pylon_lib::{ EngineStatsStreamMode, ModelDiscoveryProvider, PylonQueueMismatchRetryConfig, - PylonRetryConfig, TunnelTransportProtocol, + PylonRetryConfig, TunnelForwardingConfig, TunnelTransportProtocol, }; use reqwest::header::HeaderName; @@ -413,6 +421,23 @@ mod tests { assert!(retry.retryable_upstream_status_codes.is_empty()); } + #[test] + fn pylon_derive_dynamo_priority_cli_default_matches_runtime_default() { + let args = parse_args(""); + + assert_eq!( + args.pylon_derive_dynamo_priority, + TunnelForwardingConfig::default().derive_dynamo_priority + ); + } + + #[test] + fn pylon_derive_dynamo_priority_cli_override_is_applied() { + let args = parse_argv(&["--pylon-derive-dynamo-priority=false"]); + + assert!(!args.pylon_derive_dynamo_priority); + } + #[test] fn pylon_queue_mismatch_retry_cli_defaults_match_runtime_defaults() { let args = parse_args(""); diff --git a/src/libraries/rust/stargate/crates/pylon/src/startup.rs b/src/libraries/rust/stargate/crates/pylon/src/startup.rs index 979ef3a65..371545ab4 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/startup.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/startup.rs @@ -101,6 +101,7 @@ pub(crate) struct PylonStartupPlan { model_source: ModelSource, pylon_retry: PylonRetryConfig, queue_mismatch_retry: PylonQueueMismatchRetryConfig, + derive_dynamo_priority: bool, model_initialization: ModelInitialization, bringup: BringupConfig, request_quality_monitor: RequestQualityMonitorConfig, @@ -146,6 +147,7 @@ impl PylonStartupPlan { model_source, pylon_retry: pylon_retry_config_from_args(args)?, queue_mismatch_retry: pylon_queue_mismatch_retry_config_from_args(args)?, + derive_dynamo_priority: args.pylon_derive_dynamo_priority, model_initialization, bringup: BringupConfig { enabled: !args.disable_bringup, @@ -504,6 +506,7 @@ fn tunnel_forwarding_config_from_plan( metrics: Some(metrics), retry: plan.pylon_retry.clone(), queue_mismatch_retry: plan.queue_mismatch_retry.clone(), + derive_dynamo_priority: plan.derive_dynamo_priority, ..Default::default() } } @@ -1113,6 +1116,15 @@ mod tests { tunnel.shutdown().await; } + #[test] + fn derive_dynamo_priority_flows_from_args_to_forwarding_config() { + let (_, default_plan) = startup(&[]); + assert!(test_forwarding(&default_plan).derive_dynamo_priority); + + let (_, disabled_plan) = startup(&["--pylon-derive-dynamo-priority=false"]); + assert!(!test_forwarding(&disabled_plan).derive_dynamo_priority); + } + #[test] fn direct_tunnel_config_from_plan_preserves_runtime_inputs() { let (args, plan) = startup(&[ From 37c5ed2f83eccd47e7733992ee2eef12f2cef80f Mon Sep 17 00:00:00 2001 From: along Date: Wed, 5 Aug 2026 14:01:34 -0700 Subject: [PATCH 02/17] test(pylon): use integer strict-priority value in spoof tests Dynamo parses x-dynamo-request-strict-priority as a u32 queue tier, so the spoofed test value should look like the real contract. The strip assertion is name-based and unaffected. Signed-off-by: along --- .../stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index c06f284fe..c607a11ff 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -1705,7 +1705,7 @@ async fn quic_tunnel_derives_dynamo_priority_from_x_priority() { tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-1", "11"); headers.insert("x-priority", "7".parse().unwrap()); headers.insert("x-dynamo-request-priority", "42".parse().unwrap()); - headers.insert("x-dynamo-request-strict-priority", "true".parse().unwrap()); + headers.insert("x-dynamo-request-strict-priority", "1".parse().unwrap()); tunnel .send(headers, br#"{"messages":[],"stream":true}"#) .await; @@ -1752,7 +1752,7 @@ async fn quic_tunnel_dynamo_priority_derivation_can_be_disabled() { let mut headers = tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-3", "11"); headers.insert("x-priority", "7".parse().unwrap()); - headers.insert("x-dynamo-request-strict-priority", "true".parse().unwrap()); + headers.insert("x-dynamo-request-strict-priority", "1".parse().unwrap()); tunnel .send(headers, br#"{"messages":[],"stream":true}"#) .await; From 34775011cac09a25757ced95222b6526424b88d5 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 17:19:10 -0700 Subject: [PATCH 03/17] test(stargate): drop mock-dynamo priority-header recording The recording had no automated consumer in this PR: the tunnel tests assert receiver-side via an in-process echo backend, and live verification uses the per-request pylon log. The fixture change will land together with the e2e test that reads it. Signed-off-by: along --- .../stargate/crates/mock-dynamo/BUILD.bazel | 8 +-- .../stargate/crates/mock-dynamo/src/openai.rs | 11 +--- .../crates/mock-dynamo/src/test_control.rs | 59 ----------------- .../stargate/crates/mock-dynamo/src/tests.rs | 66 +------------------ 4 files changed, 4 insertions(+), 140 deletions(-) diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel b/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel index 68ad1f0b3..05a653501 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel +++ b/src/libraries/rust/stargate/crates/mock-dynamo/BUILD.bazel @@ -6,7 +6,7 @@ # helper only). load("@stargate_crates//:defs.bzl", "aliases", "all_crate_deps") -load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_test") +load("@rules_rust//rust:defs.bzl", "rust_binary") rust_binary( name = "mock-dynamo", @@ -17,9 +17,3 @@ rust_binary( visibility = ["//visibility:public"], deps = all_crate_deps(normal = True), ) - -rust_test( - name = "mock-dynamo_test", - crate = ":mock-dynamo", - deps = all_crate_deps(normal_dev = True), -) diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs index df718ec12..7604e2307 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/openai.rs @@ -26,9 +26,7 @@ use tracing::info; use crate::AppState; use crate::kv_cache::{KvCacheAccess, KvCacheStats, insert_kv_cache_headers}; use crate::stats_stream::StatsStreamEvent; -use crate::test_control::{ - TestEndpoint, TestRequestClass, recorded_priority_headers, request_class, -}; +use crate::test_control::{TestEndpoint, TestRequestClass, request_class}; use crate::timing::{ embedding_item_count, jitter_ms, non_streaming_delay, optional_header, prefill_delay, request_embedding_tokens, request_input_tokens, request_output_tokens, response_input_tokens, @@ -422,12 +420,7 @@ impl AppState { async fn record_request(&self, headers: &HeaderMap, endpoint: TestEndpoint, model: &str) { let request_class = request_class(headers); self.test_control - .record_request( - endpoint, - model, - request_class, - recorded_priority_headers(headers), - ) + .record_request(endpoint, model, request_class) .await; if request_class == TestRequestClass::PylonGenerated { self.test_control.wait_for_bringup_release(model).await; diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs index ad373920c..f77a23bfb 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/test_control.rs @@ -56,22 +56,12 @@ pub(crate) struct ModelTestControlUpdate { type TestCounterKey = (TestEndpoint, String, TestRequestClass); -/// Raw priority header values seen on the most recent request, kept so tunnel -/// and gateway tests can assert what actually reached the mock engine. -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] -pub(crate) struct RecordedPriorityHeaders { - pub(crate) x_priority: Option, - pub(crate) dynamo_request_priority: Option, - pub(crate) dynamo_request_strict_priority: Option, -} - #[derive(Debug, Default)] struct TestControlInner { discovered_models: BTreeSet, model_discovery_requests: u64, models: BTreeMap, counters: BTreeMap, - priority_headers: BTreeMap<(TestEndpoint, String), RecordedPriorityHeaders>, } #[derive(Debug, Clone, Default)] @@ -88,20 +78,12 @@ pub(crate) struct TestCounterSnapshot { pub(crate) count: u64, } -#[derive(Debug, Clone, Serialize)] -pub(crate) struct PriorityHeadersSnapshot { - pub(crate) endpoint: TestEndpoint, - pub(crate) model: String, - pub(crate) headers: RecordedPriorityHeaders, -} - #[derive(Debug, Clone, Serialize)] pub(crate) struct TestControlSnapshot { pub(crate) discovered_models: Vec, pub(crate) model_discovery_requests: u64, pub(crate) models: BTreeMap, pub(crate) counters: Vec, - pub(crate) priority_headers: Vec, } impl TestControlState { @@ -188,7 +170,6 @@ impl TestControlState { endpoint: TestEndpoint, model: &str, request_class: TestRequestClass, - priority_headers: RecordedPriorityHeaders, ) { let mut inner = self.inner.lock().await; let count = inner @@ -196,11 +177,6 @@ impl TestControlState { .entry((endpoint, model.to_string(), request_class)) .or_default(); *count = count.saturating_add(1); - // Last-seen semantics: a request without priority headers overwrites - // earlier values, so the snapshot always reflects the latest request. - inner - .priority_headers - .insert((endpoint, model.to_string()), priority_headers); } pub(crate) async fn snapshot(&self) -> TestControlSnapshot { @@ -221,15 +197,6 @@ impl TestControlState { }, ) .collect(), - priority_headers: inner - .priority_headers - .iter() - .map(|((endpoint, model), headers)| PriorityHeadersSnapshot { - endpoint: *endpoint, - model: model.clone(), - headers: headers.clone(), - }) - .collect(), } } } @@ -268,18 +235,6 @@ impl TestControlSnapshot { }) .map_or(0, |counter| counter.count) } - - #[cfg(test)] - pub(crate) fn priority_headers( - &self, - endpoint: TestEndpoint, - model: &str, - ) -> Option<&RecordedPriorityHeaders> { - self.priority_headers - .iter() - .find(|entry| entry.endpoint == endpoint && entry.model == model) - .map(|entry| &entry.headers) - } } pub(crate) async fn update_model_test_control( @@ -297,20 +252,6 @@ pub(crate) async fn test_control_snapshot( Json(state.test_control.snapshot().await) } -pub(crate) fn recorded_priority_headers(headers: &HeaderMap) -> RecordedPriorityHeaders { - let header_value = |name: &str| { - headers - .get(name) - .and_then(|value| value.to_str().ok()) - .map(str::to_string) - }; - RecordedPriorityHeaders { - x_priority: header_value("x-priority"), - dynamo_request_priority: header_value("x-dynamo-request-priority"), - dynamo_request_strict_priority: header_value("x-dynamo-request-strict-priority"), - } -} - pub(crate) fn request_class(headers: &HeaderMap) -> TestRequestClass { match headers .get("x-request-id") diff --git a/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs b/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs index 786e3c5c9..d4ba751a0 100644 --- a/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs +++ b/src/libraries/rust/stargate/crates/mock-dynamo/src/tests.rs @@ -191,12 +191,7 @@ async fn test_controls_count_endpoint_model_and_request_class() { ]; for (endpoint, model, request_class) in recorded { controls - .record_request( - endpoint, - model, - request_class, - RecordedPriorityHeaders::default(), - ) + .record_request(endpoint, model, request_class) .await; } @@ -313,65 +308,6 @@ async fn test_control_http_api_updates_one_model_and_reports_request_counters() server.abort(); } -#[tokio::test] -async fn test_control_snapshot_reports_last_seen_priority_headers() { - let state = test_state(); - let observed_control = state.test_control.clone(); - let app = Router::new() - .route("/v1/chat/completions", post(chat_completions)) - .route("/test-control", get(test_control_snapshot)) - .with_state(state); - let (addr, server) = spawn_test_app(app).await; - - let body = r#"{"model":"model-a","messages":[],"max_tokens":1,"stream":false}"#; - let prioritized_response = json_response( - addr, - "POST", - "/v1/chat/completions", - "connection: close\r\nx-request-id: user-8\r\nx-priority: 7\r\nx-dynamo-request-priority: 2147483640", - body, - ) - .await; - assert!(prioritized_response.starts_with("HTTP/1.1 200 OK")); - - let snapshot = observed_control.snapshot().await; - assert_eq!( - snapshot.priority_headers(TestEndpoint::ChatCompletions, "model-a"), - Some(&RecordedPriorityHeaders { - x_priority: Some("7".to_string()), - dynamo_request_priority: Some("2147483640".to_string()), - dynamo_request_strict_priority: None, - }) - ); - - let snapshot_response = raw_http_request( - addr, - &format!("GET /test-control HTTP/1.1\r\nhost: {addr}\r\nconnection: close\r\n\r\n"), - ) - .await; - assert!(snapshot_response.contains(r#""dynamo_request_priority":"2147483640""#)); - - // A later request without priority headers overwrites the recorded values, - // so the snapshot always reflects the latest request. - let plain_response = json_response( - addr, - "POST", - "/v1/chat/completions", - "connection: close\r\nx-request-id: user-9", - body, - ) - .await; - assert!(plain_response.starts_with("HTTP/1.1 200 OK")); - - let snapshot = observed_control.snapshot().await; - assert_eq!( - snapshot.priority_headers(TestEndpoint::ChatCompletions, "model-a"), - Some(&RecordedPriorityHeaders::default()) - ); - - server.abort(); -} - #[tokio::test] async fn model_discovery_http_api_returns_and_replaces_authoritative_models() { let state = test_state(); From 51d04dedc027a53ca3243b5c5a3912fbbe810990 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 17:45:16 -0700 Subject: [PATCH 04/17] feat(pylon): always emit bounded engine priority via upstream backend config Rework the engine priority contract from review feedback on the initial derivation: - Always emit x-dynamo-request-priority and x-dynamo-request-strict-priority on inference requests in dynamo mode. Dynamo resolves each field from the header when present, falling back to client-controlled request-body hints otherwise, so emitting only on configured requests left the body as an open priority channel. Unconfigured requests now carry the lowest value 0 and the strict tier is always 0. - Replace the i32::MAX anchor with max(0, ceiling - x). Dynamo reads the value as seconds of arrival-time head start, so the old anchor made every configured request outrank unconfigured traffic permanently. The ceiling is configurable (--pylon-priority-ceiling, default 3600) and named backend-neutrally because the platform priority band is platform policy. - Replace the derive_dynamo_priority bool with --pylon-upstream-backend and move all Dynamo names and the mapping into a backend module, so per-backend flags cannot accumulate and core.rs keeps no engine-specific names. The strip stays active in every mode. - Make RequiredTunnelHeaders.priority an Option and thread the validated value through, deleting the second x-priority parser the derive block previously needed. - Drop the dedicated per-derivation info log. The existing per-request observation log already carries the priority; the request span gains the inbound priority next to the derived value, startup logs the resolved backend and ceiling, and stripped engine header names log at debug. Signed-off-by: along --- .../crates/pylon-lib/src/bringup/upstream.rs | 2 +- .../rust/stargate/crates/pylon-lib/src/lib.rs | 6 +- .../crates/pylon-lib/src/queue_admission.rs | 8 +- .../crates/pylon-lib/src/quic_http_tunnel.rs | 2 + .../pylon-lib/src/quic_http_tunnel/backend.rs | 129 ++++++++++++++++++ .../pylon-lib/src/quic_http_tunnel/core.rs | 109 +++++++-------- .../src/quic_http_tunnel/raw_quic.rs | 3 +- .../pylon-lib/src/quic_http_tunnel/tests.rs | 102 ++++++++------ .../crates/pylon-lib/src/request_observer.rs | 12 +- .../pylon-lib/src/request_observer/headers.rs | 7 +- .../crates/pylon-lib/src/stats/collector.rs | 4 +- .../rust/stargate/crates/pylon/src/main.rs | 51 +++++-- .../rust/stargate/crates/pylon/src/startup.rs | 37 +++-- 13 files changed, 328 insertions(+), 144 deletions(-) create mode 100644 src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/bringup/upstream.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/bringup/upstream.rs index 0efb17cf4..077ec9f46 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/bringup/upstream.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/bringup/upstream.rs @@ -94,7 +94,7 @@ pub(super) async fn send_completion_request( request_id: request_id.clone(), routing_key: None, model_id: model_id.to_string(), - priority: 0, + priority: None, input_tokens: u64::try_from(input_tokens).unwrap_or(u64::MAX), accepted_at: std::time::Instant::now(), }, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/lib.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/lib.rs index d6b9202e4..b32f5e681 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/lib.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/lib.rs @@ -45,9 +45,9 @@ pub use model_lifecycle::{ }; pub use queue_admission::PylonQueueMismatchRetryConfig; pub use quic_http_tunnel::{ - DEFAULT_MAX_SSE_BUFFER_BYTES, PylonRetryConfig, QuicHttpTunnelConfig, QuicHttpTunnelHandle, - ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, TunnelError, TunnelForwardingConfig, - start_quic_http_tunnel, start_reverse_quic_tunnel, + DEFAULT_MAX_SSE_BUFFER_BYTES, DEFAULT_PRIORITY_CEILING, PylonRetryConfig, QuicHttpTunnelConfig, + QuicHttpTunnelHandle, ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, TunnelError, + TunnelForwardingConfig, UpstreamBackend, start_quic_http_tunnel, start_reverse_quic_tunnel, }; pub use registration::{ ClientError, InferenceServerRegistrationClient, InferenceServerRegistrationConfig, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs index eb100fa3b..b4d0960dd 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs @@ -304,7 +304,7 @@ impl LiveRequestState { }) .filter(|request| request.generation == *generation); model.queue_estimate_ms_for_priority_excluding( - required.priority, + required.priority.unwrap_or_default(), excluded_request, ) }) @@ -346,7 +346,9 @@ impl LiveRequestState { let request_id = required.request_id.clone(); let request = TrackedPromptRequest { generation, - priority: required.priority, + // Queue accounting treats unconfigured as priority 0; the + // absent-vs-0 distinction only matters to the engine derivation. + priority: required.priority.unwrap_or_default(), input_tokens: required.input_tokens, phase: TrackedPromptPhase::Pending, active_chat_output_tps: None, @@ -842,7 +844,7 @@ mod tests { request_id: request_id.to_string(), routing_key: None, model_id: model_id.to_string(), - priority, + priority: Some(priority), input_tokens, accepted_at: Instant::now(), } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel.rs index aafdaf23c..489b3174b 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel.rs @@ -13,6 +13,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +mod backend; mod core; mod endpoint; mod http3; @@ -23,6 +24,7 @@ mod server; mod tests; mod webtransport; +pub use backend::{DEFAULT_PRIORITY_CEILING, UpstreamBackend}; pub use core::{DEFAULT_MAX_SSE_BUFFER_BYTES, PylonRetryConfig, TunnelForwardingConfig}; pub use endpoint::TunnelError; pub use reverse::{ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, start_reverse_quic_tunnel}; diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs new file mode 100644 index 000000000..9a47385c1 --- /dev/null +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -0,0 +1,129 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Upstream inference-server dialects. +//! +//! Pylon presents one contract upward (the platform tunnel headers, notably +//! `x-priority`) and translates it into the dialect of the engine it fronts +//! at the last hop. The gateway and Stargate stay backend-agnostic; all +//! engine-specific names and encodings live in this module. + +use std::fmt; +use std::str::FromStr; + +/// Which engine dialect pylon speaks to its local upstream. +/// +/// One enum rather than per-backend flags: future engines add a variant and +/// a submodule here, never a new CLI flag. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum UpstreamBackend { + /// Forward requests unchanged. No engine priority headers are derived; + /// inbound engine-control headers are still stripped. + Passthrough, + /// Dynamo dialect: derive the engine priority headers from `x-priority`. + #[default] + Dynamo, +} + +impl UpstreamBackend { + pub fn as_str(&self) -> &'static str { + match self { + Self::Passthrough => "passthrough", + Self::Dynamo => "dynamo", + } + } +} + +impl fmt::Display for UpstreamBackend { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for UpstreamBackend { + type Err = String; + + fn from_str(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "passthrough" => Ok(Self::Passthrough), + "dynamo" => Ok(Self::Dynamo), + other => Err(format!( + "unknown upstream backend {other:?}; expected \"passthrough\" or \"dynamo\"" + )), + } + } +} + +/// Default seconds of scheduling head start for the most urgent platform +/// priority (`x-priority: 0`); see [`dynamo::request_priority`]. +pub const DEFAULT_PRIORITY_CEILING: u32 = 3600; + +pub(crate) mod dynamo { + use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; + + /// Engine-facing priority headers in the Dynamo contract. Pylon owns + /// this contract, so the names stay out of the shared tunnel contract. + pub(crate) const HEADER_REQUEST_PRIORITY: &str = "x-dynamo-request-priority"; + pub(crate) const HEADER_REQUEST_STRICT_PRIORITY: &str = "x-dynamo-request-strict-priority"; + const REQUEST_HEADER_PREFIX: &str = "x-dynamo-request-"; + + /// Inbound headers under the Dynamo request-priority prefix are always + /// stripped, in every backend mode: pylon is the only writer of these + /// values, so a client cannot set engine priority through them. + /// Dynamo's non-priority routing headers (worker pinning, tenant cache + /// salt) are outside this prefix and tracked separately. + pub(crate) fn is_engine_priority_header(name: &HeaderName) -> bool { + name.as_str().starts_with(REQUEST_HEADER_PREFIX) + } + + /// Map platform priority to Dynamo request priority. + /// + /// Dynamo reads the value as seconds of arrival-time head start in its + /// router queue (higher wins, i32), while `x-priority` is a rank (lower + /// wins, u32, absent = unconfigured). The mapping is + /// `max(0, ceiling - x)`, with absent treated as the lowest priority: + /// a bounded head start that queue aging can overcome, rather than a + /// permanent tier above unconfigured traffic. + pub(crate) fn request_priority(priority: Option, ceiling: u32) -> i32 { + let ceiling = ceiling.min(i32::MAX as u32); + let rank = priority.unwrap_or(ceiling).min(ceiling); + (ceiling - rank) as i32 + } + + /// Emit both Dynamo priority headers on every inference request. + /// + /// Dynamo resolves each priority field from the header when present and + /// well-formed, falling back to the client-controlled request body + /// (`nvext.agent_hints`) otherwise. Always emitting both headers makes + /// the platform the only source of engine priority: requests without a + /// platform priority carry the lowest value instead of leaving the body + /// fallback open, and the strict tier is pinned to the default. + pub(crate) fn apply_priority_headers( + priority: Option, + ceiling: u32, + upstream_headers: &mut HeaderMap, + ) -> i32 { + let dynamo_priority = request_priority(priority, ceiling); + upstream_headers.insert( + HeaderName::from_static(HEADER_REQUEST_PRIORITY), + HeaderValue::from(dynamo_priority), + ); + upstream_headers.insert( + HeaderName::from_static(HEADER_REQUEST_STRICT_PRIORITY), + HeaderValue::from_static("0"), + ); + dynamo_priority + } +} diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index cf362ca00..24a3cc36f 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -27,9 +27,8 @@ use reqwest::{Client, Error as ReqwestError, Method, Response, StatusCode}; use sonic_rs::JsonValueTrait; use stargate_protocol::common::is_hop_by_hop_header; use stargate_protocol::tunnel_contract::{ - HEADER_MODEL, HEADER_PRIORITY, HEADER_REQUEST_ID, HEADER_STARGATE_EXPECTED_QUEUE_MS, - HEADER_STARGATE_RETRY_AFTER_MS, HEADER_STARGATE_RETRY_REASON, HEADER_STARGATE_RETRYABLE, - HEADER_STARGATE_UPSTREAM_RETRYABLE, + HEADER_MODEL, HEADER_STARGATE_EXPECTED_QUEUE_MS, HEADER_STARGATE_RETRY_AFTER_MS, + HEADER_STARGATE_RETRY_REASON, HEADER_STARGATE_RETRYABLE, HEADER_STARGATE_UPSTREAM_RETRYABLE, }; use stargate_telemetry::{ inject_trace_context, parent_context_from_headers, traceparent_from_headers, @@ -38,6 +37,7 @@ use tokio_util::{sync::CancellationToken, task::TaskTracker}; use tracing::{Instrument, Span, field}; use tracing_opentelemetry::OpenTelemetrySpanExt; +use super::backend::{self, DEFAULT_PRIORITY_CEILING, UpstreamBackend}; use crate::output_token_parser::{OutputTokenParser, OutputTokenProgress}; use crate::queue_admission::{ PylonQueueMismatchRetryConfig, QueueAdmissionDecision, QueueTrackedRequestGuard, @@ -106,9 +106,12 @@ pub struct TunnelForwardingConfig { pub request_quality_monitor: RequestQualityMonitorConfig, pub retry: PylonRetryConfig, pub queue_mismatch_retry: PylonQueueMismatchRetryConfig, - /// Derive x-dynamo-request-priority for the upstream engine from x-priority. - /// Inbound x-dynamo-request-* headers are stripped regardless of this gate. - pub derive_dynamo_priority: bool, + /// Engine dialect spoken to the local upstream. Inbound engine priority + /// headers are stripped in every mode; only derivation is per-backend. + pub upstream_backend: UpstreamBackend, + /// Seconds of scheduling head start for the most urgent platform + /// priority; see [`backend::dynamo::request_priority`]. + pub priority_ceiling: u32, pub metrics: Option>, #[cfg(test)] pub webtransport_stream_header_wait_tx: Option>, @@ -125,7 +128,8 @@ impl Default for TunnelForwardingConfig { request_quality_monitor: RequestQualityMonitorConfig::default(), retry: PylonRetryConfig::default(), queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(), - derive_dynamo_priority: true, + upstream_backend: UpstreamBackend::default(), + priority_ceiling: DEFAULT_PRIORITY_CEILING, metrics: None, #[cfg(test)] webtransport_stream_header_wait_tx: None, @@ -146,7 +150,8 @@ pub(super) struct TunnelServerApp { pub(super) request_quality_monitor: RequestQualityMonitorConfig, pub(super) retry: PylonRetryConfig, pub(super) queue_mismatch_retry: PylonQueueMismatchRetryConfig, - pub(super) derive_dynamo_priority: bool, + pub(super) upstream_backend: UpstreamBackend, + pub(super) priority_ceiling: u32, pub(super) metrics: Option>, #[cfg(test)] pub(super) webtransport_stream_header_wait_tx: Option>, @@ -170,7 +175,8 @@ impl TunnelServerApp { request_quality_monitor: forwarding.request_quality_monitor, retry: forwarding.retry, queue_mismatch_retry: forwarding.queue_mismatch_retry, - derive_dynamo_priority: forwarding.derive_dynamo_priority, + upstream_backend: forwarding.upstream_backend, + priority_ceiling: forwarding.priority_ceiling, metrics: forwarding.metrics, #[cfg(test)] webtransport_stream_header_wait_tx: forwarding.webtransport_stream_header_wait_tx, @@ -661,13 +667,18 @@ pub(super) async fn forward_tunnel_request( } } + // None for health requests, which skip header validation and are not + // client inference traffic; nothing to trace or derive for them. + let upstream_context = lifecycle.as_ref().map(|lifecycle| UpstreamRequestContext { + priority: lifecycle.required.priority, + }); let response = match send_upstream_request( app, method, &path_and_query, &request_headers, body_bytes, - !health_request, + upstream_context, ) .await { @@ -700,15 +711,24 @@ pub(super) async fn forward_tunnel_request( Ok(()) } +/// Fields of the validated tunnel headers the upstream send path needs. +/// `None` at the call site means a health request: unvalidated, untraced, +/// and never carrying derived engine headers. +#[derive(Clone, Copy)] +struct UpstreamRequestContext { + /// Platform priority; `None` when the request carried no x-priority. + priority: Option, +} + async fn send_upstream_request( app: &TunnelServerApp, method: Method, path_and_query: &str, request_headers: &HeaderMap, body_bytes: Vec, - traced: bool, + context: Option, ) -> Result { - let span = if traced { + let span = if context.is_some() { let span = tracing::info_span!( "pylon_upstream_http_request", otel_parent = field::Empty, @@ -717,6 +737,7 @@ async fn send_upstream_request( inference_server.id = %app.inference_server_id, upstream.status = field::Empty, upstream.error = field::Empty, + priority = field::Empty, dynamo.request_priority = field::Empty, ); let _ = span.set_parent(pylon_upstream_parent_context(request_headers)); @@ -731,30 +752,23 @@ async fn send_upstream_request( for (name, value) in request_headers { if should_forward_header(name, &app.retry) { upstream_headers.append(name, value.clone()); + } else if backend::dynamo::is_engine_priority_header(name) { + // Values are client-controlled; log the name only. + tracing::debug!(header = %name, "stripped inbound engine priority header"); } } - // `traced` excludes health requests, which skip header validation and are - // not client inference traffic; nothing to derive for them. - if app.derive_dynamo_priority && traced { - if let Some(priority) = x_priority_header_value(request_headers) { - let dynamo_priority = dynamo_request_priority(priority); - upstream_headers.insert( - HeaderName::from_static(HEADER_DYNAMO_REQUEST_PRIORITY), - HeaderValue::from(dynamo_priority), + if let Some(context) = context { + if let Some(priority) = context.priority { + span.record("priority", priority); + } + if app.upstream_backend == UpstreamBackend::Dynamo { + let dynamo_priority = backend::dynamo::apply_priority_headers( + context.priority, + app.priority_ceiling, + &mut upstream_headers, ); span.record("dynamo.request_priority", dynamo_priority); - tracing::info!( - request_id = request_headers - .get(HEADER_REQUEST_ID) - .and_then(|value| value.to_str().ok()) - .unwrap_or_default(), - priority, - dynamo.request_priority = dynamo_priority, - "derived dynamo request priority" - ); } - } - if traced { inject_trace_context(&mut upstream_headers, &span.context()); } let send = async { @@ -1087,44 +1101,13 @@ pub(super) fn join_base_path(base: &str, path_and_query: &str) -> Result bool { !is_tunnel_control_header(name, retry) - && !is_dynamo_request_header(name) + && !backend::dynamo::is_engine_priority_header(name) && !matches!( name.as_str(), "host" | "x-method" | "x-path" | HEADER_STARGATE_EXPECTED_QUEUE_MS ) } -/// Engine-facing priority header in the Dynamo contract; Pylon owns this -/// contract, so the constant stays out of the shared tunnel contract. -pub(super) const HEADER_DYNAMO_REQUEST_PRIORITY: &str = "x-dynamo-request-priority"; -const DYNAMO_REQUEST_HEADER_PREFIX: &str = "x-dynamo-request-"; - -/// Client-supplied Dynamo request headers never reach the engine; Pylon is -/// the only writer of x-dynamo-request-* values. -pub(super) fn is_dynamo_request_header(name: &HeaderName) -> bool { - name.as_str().starts_with(DYNAMO_REQUEST_HEADER_PREFIX) -} - -/// Dynamo schedules on an i32 where higher wins and silently drops values -/// that do not parse as i32, while x-priority is a u32 where lower wins, so -/// the mapping inverts and clamps to keep every configured priority valid. -pub(super) fn dynamo_request_priority(priority: u32) -> i32 { - i32::MAX - i32::try_from(priority).unwrap_or(i32::MAX) -} - -/// Absent stays absent: a request without x-priority must not be promoted to -/// maximum engine priority, so this reads the raw header instead of the -/// parsed default of 0. Malformed values were already rejected upstream. -pub(super) fn x_priority_header_value(headers: &HeaderMap) -> Option { - headers - .get(HEADER_PRIORITY)? - .to_str() - .ok()? - .trim() - .parse() - .ok() -} - pub(super) fn should_forward_response_header(name: &HeaderName, retry: &PylonRetryConfig) -> bool { !is_tunnel_control_header(name, retry) && name != CONTENT_LENGTH } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs index 725518bfd..e39d81d49 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs @@ -171,7 +171,8 @@ mod tests { request_quality_monitor: RequestQualityMonitorConfig::default(), retry: PylonRetryConfig::default(), queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(), - derive_dynamo_priority: true, + upstream_backend: crate::quic_http_tunnel::UpstreamBackend::default(), + priority_ceiling: crate::quic_http_tunnel::DEFAULT_PRIORITY_CEILING, metrics: None, #[cfg(test)] webtransport_stream_header_wait_tx: None, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index c607a11ff..9796f8003 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -13,11 +13,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::backend::{DEFAULT_PRIORITY_CEILING, UpstreamBackend, dynamo}; use super::core::{ - MAX_SPECULATIVE_REQUEST_BODY_PREALLOC_BYTES, TunnelServerApp, dynamo_request_priority, - extend_body_from_buf, is_health_request_path, otel_parent_from_headers, - pylon_upstream_parent_context, request_body_buffer, request_body_capacity, - should_forward_header, should_forward_response_header, x_priority_header_value, + MAX_SPECULATIVE_REQUEST_BODY_PREALLOC_BYTES, TunnelServerApp, extend_body_from_buf, + is_health_request_path, otel_parent_from_headers, pylon_upstream_parent_context, + request_body_buffer, request_body_capacity, should_forward_header, + should_forward_response_header, }; use super::endpoint::{ build_trusted_client_config, derive_sni, make_server_config, target_authority, @@ -499,27 +500,41 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() } #[test] -fn pylon_dynamo_request_priority_inverts_and_clamps_to_i32() { - assert_eq!(dynamo_request_priority(0), i32::MAX); - assert_eq!(dynamo_request_priority(7), i32::MAX - 7); - assert_eq!(dynamo_request_priority(i32::MAX as u32), 0); - assert_eq!(dynamo_request_priority(i32::MAX as u32 + 1), 0); - assert_eq!(dynamo_request_priority(u32::MAX), 0); +fn pylon_dynamo_request_priority_inverts_within_bounded_ceiling() { + let ceiling = DEFAULT_PRIORITY_CEILING; + // Most urgent platform rank gets the full head start. + assert_eq!(dynamo::request_priority(Some(0), ceiling), ceiling as i32); + assert_eq!( + dynamo::request_priority(Some(7), ceiling), + (ceiling - 7) as i32 + ); + // Unconfigured and beyond-ceiling ranks both land at the lowest value. + assert_eq!(dynamo::request_priority(None, ceiling), 0); + assert_eq!(dynamo::request_priority(Some(ceiling), ceiling), 0); + assert_eq!(dynamo::request_priority(Some(u32::MAX), ceiling), 0); + // A ceiling beyond i32 is clamped so the emitted value stays a valid i32. + assert_eq!(dynamo::request_priority(Some(0), u32::MAX), i32::MAX); + assert_eq!(dynamo::request_priority(None, u32::MAX), 0); } #[test] -fn pylon_x_priority_header_value_distinguishes_absent_from_zero() { +fn pylon_dynamo_priority_headers_are_always_emitted() { let mut headers = HeaderMap::new(); - assert_eq!(x_priority_header_value(&headers), None); - - headers.insert("x-priority", "0".parse().unwrap()); - assert_eq!(x_priority_header_value(&headers), Some(0)); - - headers.insert("x-priority", " 7 ".parse().unwrap()); - assert_eq!(x_priority_header_value(&headers), Some(7)); + let emitted = dynamo::apply_priority_headers(Some(7), DEFAULT_PRIORITY_CEILING, &mut headers); + assert_eq!(emitted, (DEFAULT_PRIORITY_CEILING - 7) as i32); + assert_eq!( + headers["x-dynamo-request-priority"], + emitted.to_string().as_str() + ); + assert_eq!(headers["x-dynamo-request-strict-priority"], "0"); - headers.insert("x-priority", "not-a-priority".parse().unwrap()); - assert_eq!(x_priority_header_value(&headers), None); + // Absent platform priority pins both headers to the lowest values so the + // engine never falls back to client-controlled body hints. + let mut headers = HeaderMap::new(); + let emitted = dynamo::apply_priority_headers(None, DEFAULT_PRIORITY_CEILING, &mut headers); + assert_eq!(emitted, 0); + assert_eq!(headers["x-dynamo-request-priority"], "0"); + assert_eq!(headers["x-dynamo-request-strict-priority"], "0"); } #[test] @@ -794,7 +809,7 @@ async fn start_queue_mismatch_test_tunnel( request_id: "req-already-queued".to_string(), routing_key: Some("rk-1".to_string()), model_id: "model-a".to_string(), - priority: 0, + priority: None, input_tokens: 100, accepted_at: std::time::Instant::now(), }); @@ -1662,19 +1677,21 @@ async fn quic_tunnel_forwards_to_http_backend() { tunnel.shutdown().await; } +/// Echoes the Dynamo priority headers the backend received, so the tunnel +/// tests assert on what actually crossed the pylon-to-engine hop. fn dynamo_priority_echo_router() -> Router { Router::new().route( "/v1/chat/completions", post(|req: Request| async move { - let dynamo_priority = req - .headers() - .get("x-dynamo-request-priority") - .and_then(|value| value.to_str().ok()) - .unwrap_or("absent") - .to_string(); - let saw_spoofed_dynamo_header = req - .headers() - .contains_key("x-dynamo-request-strict-priority"); + let echo_header = |name: &str| { + req.headers() + .get(name) + .and_then(|value| value.to_str().ok()) + .unwrap_or("absent") + .to_string() + }; + let dynamo_priority = echo_header("x-dynamo-request-priority"); + let dynamo_strict_priority = echo_header("x-dynamo-request-strict-priority"); let mut sse = axum::response::Sse::new(async_stream::stream! { yield Ok::<_, std::convert::Infallible>( Event::default().data(r#"{"object":"chat.completion.chunk","choices":[{"delta":{"content":"ok"}}]}"#) @@ -1687,8 +1704,8 @@ fn dynamo_priority_echo_router() -> Router { HeaderValue::from_str(&dynamo_priority).unwrap(), ); sse.headers_mut().insert( - HeaderName::from_static("x-saw-spoofed-dynamo-header"), - HeaderValue::from_str(&saw_spoofed_dynamo_header.to_string()).unwrap(), + HeaderName::from_static("x-echo-dynamo-strict-priority"), + HeaderValue::from_str(&dynamo_strict_priority).unwrap(), ); *sse.status_mut() = StatusCode::OK; sse @@ -1699,11 +1716,13 @@ fn dynamo_priority_echo_router() -> Router { #[tokio::test] async fn quic_tunnel_derives_dynamo_priority_from_x_priority() { let (config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; + let ceiling = config.forwarding.priority_ceiling; let mut tunnel = RawTunnelTest::start(config).await; let mut headers = tunnel_request_headers("/v1/chat/completions", "model-a", "req-dynamo-1", "11"); headers.insert("x-priority", "7".parse().unwrap()); + // Spoofed engine headers must be replaced by pylon-derived values. headers.insert("x-dynamo-request-priority", "42".parse().unwrap()); headers.insert("x-dynamo-request-strict-priority", "1".parse().unwrap()); tunnel @@ -1717,15 +1736,15 @@ async fn quic_tunnel_derives_dynamo_priority_from_x_priority() { .unwrap() .to_str() .unwrap(), - (i32::MAX - 7).to_string() + (ceiling - 7).to_string() ); - assert_eq!(response_headers["x-saw-spoofed-dynamo-header"], "false"); + assert_eq!(response_headers["x-echo-dynamo-strict-priority"], "0"); tunnel.shutdown().await; } #[tokio::test] -async fn quic_tunnel_omits_dynamo_priority_without_x_priority() { +async fn quic_tunnel_emits_lowest_dynamo_priority_without_x_priority() { let (config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; let mut tunnel = RawTunnelTest::start(config).await; @@ -1736,17 +1755,20 @@ async fn quic_tunnel_omits_dynamo_priority_without_x_priority() { .send(headers, br#"{"messages":[],"stream":true}"#) .await; + // Unconfigured requests carry the lowest priority instead of no header, + // so the engine never falls back to client-controlled body values. let response_headers = tunnel.response_head(StatusCode::OK).await; - assert_eq!(response_headers["x-echo-dynamo-priority"], "absent"); + assert_eq!(response_headers["x-echo-dynamo-priority"], "0"); + assert_eq!(response_headers["x-echo-dynamo-strict-priority"], "0"); tunnel.shutdown().await; } #[tokio::test] -async fn quic_tunnel_dynamo_priority_derivation_can_be_disabled() { +async fn quic_tunnel_passthrough_backend_strips_but_derives_nothing() { let (mut config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; - config.forwarding.derive_dynamo_priority = false; + config.forwarding.upstream_backend = UpstreamBackend::Passthrough; let mut tunnel = RawTunnelTest::start(config).await; let mut headers = @@ -1759,8 +1781,8 @@ async fn quic_tunnel_dynamo_priority_derivation_can_be_disabled() { let response_headers = tunnel.response_head(StatusCode::OK).await; assert_eq!(response_headers["x-echo-dynamo-priority"], "absent"); - // Stripping inbound x-dynamo-request-* headers is not gated by the flag. - assert_eq!(response_headers["x-saw-spoofed-dynamo-header"], "false"); + // Stripping inbound engine priority headers is not gated by the backend. + assert_eq!(response_headers["x-echo-dynamo-strict-priority"], "absent"); tunnel.shutdown().await; } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs index 69d0ebc31..8f559b0b7 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs @@ -175,7 +175,7 @@ impl RequestObserver { started_at: accepted_at, routing_key, model_id, - priority, + priority: priority.unwrap_or_default(), input_tokens, generation, embedding_items: None, @@ -605,14 +605,16 @@ mod tests { assert_eq!(required.routing_key.as_deref(), Some("rk-1")); assert_eq!(required.model_id, "model-a"); assert_eq!(required.input_tokens, 42); - assert_eq!(required.priority, 7); + assert_eq!(required.priority, Some(7)); } #[test] - fn validate_required_tunnel_headers_defaults_missing_priority_to_zero() { + fn validate_required_tunnel_headers_keeps_missing_priority_absent() { let required = validate_required_tunnel_headers(&request_headers("req-1", 42)).unwrap(); - assert_eq!(required.priority, 0); + // Absent stays absent rather than defaulting to 0: the engine + // priority derivation treats unconfigured and rank 0 differently. + assert_eq!(required.priority, None); } #[test] @@ -676,7 +678,7 @@ mod tests { request_id: "req-embeddings-terminal".to_string(), routing_key: Some("rk-1".to_string()), model_id: "model-embed".to_string(), - priority: 0, + priority: None, input_tokens: 12, accepted_at: Instant::now(), } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs index e3e50e73c..cc07559a2 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs @@ -62,7 +62,9 @@ pub(crate) struct RequiredTunnelHeaders { pub request_id: String, pub routing_key: Option, pub model_id: String, - pub priority: u32, + /// `None` when the request carried no x-priority header. The distinction + /// from an explicit 0 matters to the engine priority derivation. + pub priority: Option, pub input_tokens: u64, pub(crate) accepted_at: Instant, } @@ -77,8 +79,7 @@ pub(crate) fn validate_required_tunnel_headers( .ok_or_else(|| MissingRequiredHeaderError::new(HEADER_MODEL))?; let input_tokens = parse_optional_numeric_header(request_headers, HEADER_INPUT_TOKENS)? .ok_or_else(|| MissingRequiredHeaderError::new(HEADER_INPUT_TOKENS))?; - let priority = - parse_optional_numeric_header(request_headers, HEADER_PRIORITY)?.unwrap_or_default(); + let priority = parse_optional_numeric_header(request_headers, HEADER_PRIORITY)?; Ok(RequiredTunnelHeaders { request_id, routing_key, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/stats/collector.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/stats/collector.rs index 247db3288..1c77c6958 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/stats/collector.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/stats/collector.rs @@ -2048,7 +2048,7 @@ mod tests { request_id: "req-queued-after-fallback-samples".to_string(), routing_key: None, model_id: "model-a".to_string(), - priority: 0, + priority: None, input_tokens: 50, accepted_at: std::time::Instant::now(), }); @@ -2479,7 +2479,7 @@ mod tests { request_id: "req-queued".to_string(), routing_key: None, model_id: "model-a".to_string(), - priority: 0, + priority: None, input_tokens: 32, accepted_at: std::time::Instant::now(), }, diff --git a/src/libraries/rust/stargate/crates/pylon/src/main.rs b/src/libraries/rust/stargate/crates/pylon/src/main.rs index 94b5db966..29718ea57 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/main.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/main.rs @@ -14,7 +14,9 @@ // limitations under the License. use anyhow::Result; -use pylon_lib::{EngineStatsStreamMode, ModelDiscoveryProvider, TunnelTransportProtocol}; +use pylon_lib::{ + EngineStatsStreamMode, ModelDiscoveryProvider, TunnelTransportProtocol, UpstreamBackend, +}; use stargate_protocol::BackendConnectivity; use stargate_protocol::tunnel_contract::HEADER_STARGATE_UPSTREAM_RETRYABLE; @@ -204,14 +206,25 @@ struct Args { /// Optional retry-after hint in milliseconds for local queue-mismatch retries #[arg(long, env = "PYLON_QUEUE_MISMATCH_RETRY_AFTER_MS", value_name = "MS")] pylon_queue_mismatch_retry_after_ms: Option, - /// Derive x-dynamo-request-priority for the upstream engine from x-priority + /// Engine dialect spoken to the local upstream: "dynamo" derives the + /// engine priority headers from x-priority, "passthrough" derives nothing. + /// Inbound engine priority headers are stripped in every mode. #[arg( long, - action = clap::ArgAction::Set, - default_value_t = true, - env = "PYLON_DERIVE_DYNAMO_PRIORITY" + default_value = "dynamo", + env = "PYLON_UPSTREAM_BACKEND", + value_name = "BACKEND" + )] + pylon_upstream_backend: UpstreamBackend, + /// Seconds of engine scheduling head start for the most urgent platform + /// priority (x-priority: 0); lower x-priority ranks get proportionally less + #[arg( + long, + default_value_t = pylon_lib::DEFAULT_PRIORITY_CEILING, + env = "PYLON_PRIORITY_CEILING", + value_name = "SECONDS" )] - pylon_derive_dynamo_priority: bool, + pylon_priority_ceiling: u32, /// Collect post-stream output quality metrics (gibberish checks) #[arg(long, default_value_t = false)] collect_quality_metrics: bool, @@ -422,20 +435,30 @@ mod tests { } #[test] - fn pylon_derive_dynamo_priority_cli_default_matches_runtime_default() { + fn pylon_upstream_backend_cli_defaults_match_runtime_defaults() { let args = parse_args(""); + let defaults = TunnelForwardingConfig::default(); - assert_eq!( - args.pylon_derive_dynamo_priority, - TunnelForwardingConfig::default().derive_dynamo_priority - ); + assert_eq!(args.pylon_upstream_backend, defaults.upstream_backend); + assert_eq!(args.pylon_priority_ceiling, defaults.priority_ceiling); } #[test] - fn pylon_derive_dynamo_priority_cli_override_is_applied() { - let args = parse_argv(&["--pylon-derive-dynamo-priority=false"]); + fn pylon_upstream_backend_cli_overrides_are_applied() { + let args = parse_argv(&[ + "--pylon-upstream-backend", + "passthrough", + "--pylon-priority-ceiling", + "600", + ]); - assert!(!args.pylon_derive_dynamo_priority); + assert_eq!(args.pylon_upstream_backend, UpstreamBackend::Passthrough); + assert_eq!(args.pylon_priority_ceiling, 600); + } + + #[test] + fn pylon_upstream_backend_cli_rejects_unknown_backend() { + assert!(try_parse_argv(&["--pylon-upstream-backend", "sglang"]).is_err()); } #[test] diff --git a/src/libraries/rust/stargate/crates/pylon/src/startup.rs b/src/libraries/rust/stargate/crates/pylon/src/startup.rs index 371545ab4..7338f84a9 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/startup.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/startup.rs @@ -28,8 +28,8 @@ use pylon_lib::{ ModelInitialization, ModelLifecycleConfig, ModelLifecycleHandle, ModelSource, PylonMetrics, PylonQueueMismatchRetryConfig, PylonRetryConfig, PylonRuntimeState, QuicHttpTunnelConfig, QuicHttpTunnelHandle, RequestQualityMonitorConfig, StatsCollectorConfig, StatsCollectorHandle, - TunnelForwardingConfig, start_engine_stats_stream, start_metrics_server, start_model_lifecycle, - start_quic_http_tunnel, start_stats_collector_with_engine_stats, + TunnelForwardingConfig, UpstreamBackend, start_engine_stats_stream, start_metrics_server, + start_model_lifecycle, start_quic_http_tunnel, start_stats_collector_with_engine_stats, stats_aggregator_update_channel, }; use reqwest::header::HeaderName; @@ -79,6 +79,8 @@ fn log_startup_complete( inference_server_id, cluster_id = %plan.cluster_id, upstream = %plan.upstream, + upstream_backend = %plan.upstream_backend, + priority_ceiling = plan.priority_ceiling, model_ids = ?model_ids, "pylon startup complete; stargate registration started (reverse tunnel mode)" ); @@ -89,6 +91,8 @@ fn log_startup_complete( cluster_id = %plan.cluster_id, inference_server_url = registration_inference_server_url, upstream = %plan.upstream, + upstream_backend = %plan.upstream_backend, + priority_ceiling = plan.priority_ceiling, model_ids = ?model_ids, "pylon startup complete; stargate registration started (direct tunnel mode)" ); @@ -101,7 +105,8 @@ pub(crate) struct PylonStartupPlan { model_source: ModelSource, pylon_retry: PylonRetryConfig, queue_mismatch_retry: PylonQueueMismatchRetryConfig, - derive_dynamo_priority: bool, + upstream_backend: UpstreamBackend, + priority_ceiling: u32, model_initialization: ModelInitialization, bringup: BringupConfig, request_quality_monitor: RequestQualityMonitorConfig, @@ -147,7 +152,8 @@ impl PylonStartupPlan { model_source, pylon_retry: pylon_retry_config_from_args(args)?, queue_mismatch_retry: pylon_queue_mismatch_retry_config_from_args(args)?, - derive_dynamo_priority: args.pylon_derive_dynamo_priority, + upstream_backend: args.pylon_upstream_backend, + priority_ceiling: args.pylon_priority_ceiling, model_initialization, bringup: BringupConfig { enabled: !args.disable_bringup, @@ -506,7 +512,8 @@ fn tunnel_forwarding_config_from_plan( metrics: Some(metrics), retry: plan.pylon_retry.clone(), queue_mismatch_retry: plan.queue_mismatch_retry.clone(), - derive_dynamo_priority: plan.derive_dynamo_priority, + upstream_backend: plan.upstream_backend, + priority_ceiling: plan.priority_ceiling, ..Default::default() } } @@ -1117,12 +1124,24 @@ mod tests { } #[test] - fn derive_dynamo_priority_flows_from_args_to_forwarding_config() { + fn upstream_backend_flows_from_args_to_forwarding_config() { let (_, default_plan) = startup(&[]); - assert!(test_forwarding(&default_plan).derive_dynamo_priority); + let forwarding = test_forwarding(&default_plan); + assert_eq!(forwarding.upstream_backend, UpstreamBackend::Dynamo); + assert_eq!( + forwarding.priority_ceiling, + pylon_lib::DEFAULT_PRIORITY_CEILING + ); - let (_, disabled_plan) = startup(&["--pylon-derive-dynamo-priority=false"]); - assert!(!test_forwarding(&disabled_plan).derive_dynamo_priority); + let (_, passthrough_plan) = startup(&[ + "--pylon-upstream-backend", + "passthrough", + "--pylon-priority-ceiling", + "600", + ]); + let forwarding = test_forwarding(&passthrough_plan); + assert_eq!(forwarding.upstream_backend, UpstreamBackend::Passthrough); + assert_eq!(forwarding.priority_ceiling, 600); } #[test] From e851ac948c5d645262f7b6bf73cca73d119207d2 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 17:45:23 -0700 Subject: [PATCH 05/17] docs(pylon): document engine priority headers and backend flags Add the x-dynamo-request-* strip/derive rules to the gateway contract's internal headers section and the upstream backend and priority ceiling flags to the request-router operator doc. Signed-off-by: along --- docs/user/llm-request-router-load-balancing.md | 16 ++++++++++++++++ .../rust/stargate/docs/api-gateway-contract.md | 12 +++++++++++- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/docs/user/llm-request-router-load-balancing.md b/docs/user/llm-request-router-load-balancing.md index fbe0c433b..ec9482b18 100644 --- a/docs/user/llm-request-router-load-balancing.md +++ b/docs/user/llm-request-router-load-balancing.md @@ -133,6 +133,22 @@ Stargate returns HTTP `400` for a blank, unknown, or configured-but-unavailable `x-routing-method`. It also returns HTTP `400` when a required router header is missing or a numeric header is invalid. +### Engine priority translation in pylon + +Pylon translates the platform `x-priority` header into the engine's own +priority contract at the last hop. Two pylon flags control this: + +| Flag | Environment variable | Default | Meaning | +| --- | --- | --- | --- | +| `--pylon-upstream-backend` | `PYLON_UPSTREAM_BACKEND` | `dynamo` | Engine dialect. `dynamo` derives `x-dynamo-request-priority` and `x-dynamo-request-strict-priority`; `passthrough` derives nothing. | +| `--pylon-priority-ceiling` | `PYLON_PRIORITY_CEILING` | `3600` | Seconds of engine scheduling head start for `x-priority: 0`. Lower-urgency ranks get proportionally less; ranks at or beyond the ceiling and requests without `x-priority` get `0`. | + +In `dynamo` mode pylon emits both engine headers on every inference request, +so the engine never reads client-supplied priority values from headers or +request bodies. Inbound `x-dynamo-request-*` headers are stripped in every +mode. Pylon logs the resolved backend and ceiling at startup and records the +inbound and derived priority on each request log and span. + ## Apply and roll out Render the chart before applying it: diff --git a/src/libraries/rust/stargate/docs/api-gateway-contract.md b/src/libraries/rust/stargate/docs/api-gateway-contract.md index 48889ceb9..cc2afc3df 100644 --- a/src/libraries/rust/stargate/docs/api-gateway-contract.md +++ b/src/libraries/rust/stargate/docs/api-gateway-contract.md @@ -144,10 +144,20 @@ Optional trusted headers: The gateway must synthesize or validate these headers. Do not pass public caller-supplied routing headers through blindly. -Internal header: +Internal headers: - `x-stargate-expected-queue-ms`: Stargate-to-pylon only. Stargate strips caller values; pylon strips it before upstream forwarding. +- `x-dynamo-request-*` (notably `x-dynamo-request-priority` and + `x-dynamo-request-strict-priority`): pylon-to-engine only. Pylon strips + inbound values in every backend mode, so pylon is the only writer of these + headers. When pylon runs with `--pylon-upstream-backend dynamo` (the + default), it emits both headers on every inference request: the priority is + derived from `x-priority` as `max(0, ceiling - x)` with a configurable + ceiling (`--pylon-priority-ceiling`, default 3600), requests without + `x-priority` carry the lowest value `0`, and the strict tier is always `0`. + Always emitting means the engine never falls back to client-controlled + request-body priority hints. Body rules: From b2fa23687137cca488f8d0577903a80e3ae731ea Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 18:43:45 -0700 Subject: [PATCH 06/17] refactor(pylon): tighten priority plumbing and keep docs in the codebase - Revert the user-doc addition; the feature is not QA validated yet, so the flag and header documentation stays in the stargate contract doc. - Centralize the unconfigured-counts-as-0 rule in RequiredTunnelHeaders::queue_priority() instead of scattered unwrap_or_default() calls. - Rename is_engine_priority_header to is_engine_request_header: the strip covers the whole pylon-owned request-header prefix, not only priority. - Rename the send-path context to ValidatedRequestContext/validated so the health-request distinction reads off the signature. - Trim comments to stay component-focused and concise. Signed-off-by: along --- .../user/llm-request-router-load-balancing.md | 16 ------ .../crates/pylon-lib/src/queue_admission.rs | 6 +-- .../pylon-lib/src/quic_http_tunnel/backend.rs | 54 +++++++------------ .../pylon-lib/src/quic_http_tunnel/core.rs | 22 ++++---- .../pylon-lib/src/quic_http_tunnel/tests.rs | 6 +-- .../crates/pylon-lib/src/request_observer.rs | 5 +- .../pylon-lib/src/request_observer/headers.rs | 12 ++++- .../stargate/docs/api-gateway-contract.md | 4 +- 8 files changed, 49 insertions(+), 76 deletions(-) diff --git a/docs/user/llm-request-router-load-balancing.md b/docs/user/llm-request-router-load-balancing.md index ec9482b18..fbe0c433b 100644 --- a/docs/user/llm-request-router-load-balancing.md +++ b/docs/user/llm-request-router-load-balancing.md @@ -133,22 +133,6 @@ Stargate returns HTTP `400` for a blank, unknown, or configured-but-unavailable `x-routing-method`. It also returns HTTP `400` when a required router header is missing or a numeric header is invalid. -### Engine priority translation in pylon - -Pylon translates the platform `x-priority` header into the engine's own -priority contract at the last hop. Two pylon flags control this: - -| Flag | Environment variable | Default | Meaning | -| --- | --- | --- | --- | -| `--pylon-upstream-backend` | `PYLON_UPSTREAM_BACKEND` | `dynamo` | Engine dialect. `dynamo` derives `x-dynamo-request-priority` and `x-dynamo-request-strict-priority`; `passthrough` derives nothing. | -| `--pylon-priority-ceiling` | `PYLON_PRIORITY_CEILING` | `3600` | Seconds of engine scheduling head start for `x-priority: 0`. Lower-urgency ranks get proportionally less; ranks at or beyond the ceiling and requests without `x-priority` get `0`. | - -In `dynamo` mode pylon emits both engine headers on every inference request, -so the engine never reads client-supplied priority values from headers or -request bodies. Inbound `x-dynamo-request-*` headers are stripped in every -mode. Pylon logs the resolved backend and ceiling at startup and records the -inbound and derived priority on each request log and span. - ## Apply and roll out Render the chart before applying it: diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs index b4d0960dd..0a3697abe 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs @@ -304,7 +304,7 @@ impl LiveRequestState { }) .filter(|request| request.generation == *generation); model.queue_estimate_ms_for_priority_excluding( - required.priority.unwrap_or_default(), + required.queue_priority(), excluded_request, ) }) @@ -346,9 +346,7 @@ impl LiveRequestState { let request_id = required.request_id.clone(); let request = TrackedPromptRequest { generation, - // Queue accounting treats unconfigured as priority 0; the - // absent-vs-0 distinction only matters to the engine derivation. - priority: required.priority.unwrap_or_default(), + priority: required.queue_priority(), input_tokens: required.input_tokens, phase: TrackedPromptPhase::Pending, active_chat_output_tps: None, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index 9a47385c1..152367b69 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -15,24 +15,21 @@ //! Upstream inference-server dialects. //! -//! Pylon presents one contract upward (the platform tunnel headers, notably -//! `x-priority`) and translates it into the dialect of the engine it fronts -//! at the last hop. The gateway and Stargate stay backend-agnostic; all -//! engine-specific names and encodings live in this module. +//! Pylon speaks the platform tunnel contract upward and translates it into +//! the dialect of the engine it fronts. All engine-specific header names and +//! encodings live in this module. use std::fmt; use std::str::FromStr; -/// Which engine dialect pylon speaks to its local upstream. -/// -/// One enum rather than per-backend flags: future engines add a variant and -/// a submodule here, never a new CLI flag. +/// Which engine dialect pylon speaks to its local upstream. Future engines +/// add a variant and a submodule here, never a new CLI flag. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum UpstreamBackend { - /// Forward requests unchanged. No engine priority headers are derived; - /// inbound engine-control headers are still stripped. + /// Forward requests unchanged; derive nothing. Inbound engine request + /// headers are still stripped. Passthrough, - /// Dynamo dialect: derive the engine priority headers from `x-priority`. + /// Derive the engine priority headers from `x-priority`. #[default] Dynamo, } @@ -73,43 +70,30 @@ pub const DEFAULT_PRIORITY_CEILING: u32 = 3600; pub(crate) mod dynamo { use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; - /// Engine-facing priority headers in the Dynamo contract. Pylon owns - /// this contract, so the names stay out of the shared tunnel contract. + /// Engine priority headers pylon derives; the names stay out of the + /// shared tunnel contract because only pylon speaks them. pub(crate) const HEADER_REQUEST_PRIORITY: &str = "x-dynamo-request-priority"; pub(crate) const HEADER_REQUEST_STRICT_PRIORITY: &str = "x-dynamo-request-strict-priority"; const REQUEST_HEADER_PREFIX: &str = "x-dynamo-request-"; - /// Inbound headers under the Dynamo request-priority prefix are always - /// stripped, in every backend mode: pylon is the only writer of these - /// values, so a client cannot set engine priority through them. - /// Dynamo's non-priority routing headers (worker pinning, tenant cache - /// salt) are outside this prefix and tracked separately. - pub(crate) fn is_engine_priority_header(name: &HeaderName) -> bool { + /// Pylon is the only writer of headers under this prefix, so inbound + /// values are stripped in every backend mode. + pub(crate) fn is_engine_request_header(name: &HeaderName) -> bool { name.as_str().starts_with(REQUEST_HEADER_PREFIX) } - /// Map platform priority to Dynamo request priority. - /// - /// Dynamo reads the value as seconds of arrival-time head start in its - /// router queue (higher wins, i32), while `x-priority` is a rank (lower - /// wins, u32, absent = unconfigured). The mapping is - /// `max(0, ceiling - x)`, with absent treated as the lowest priority: - /// a bounded head start that queue aging can overcome, rather than a - /// permanent tier above unconfigured traffic. + /// Map the platform rank (lower wins, absent = unconfigured) to the + /// engine value (higher wins, read as seconds of queue head start): + /// `max(0, ceiling - rank)`, with absent as the lowest value. The head + /// start is bounded so prioritized traffic cannot starve the rest. pub(crate) fn request_priority(priority: Option, ceiling: u32) -> i32 { let ceiling = ceiling.min(i32::MAX as u32); let rank = priority.unwrap_or(ceiling).min(ceiling); (ceiling - rank) as i32 } - /// Emit both Dynamo priority headers on every inference request. - /// - /// Dynamo resolves each priority field from the header when present and - /// well-formed, falling back to the client-controlled request body - /// (`nvext.agent_hints`) otherwise. Always emitting both headers makes - /// the platform the only source of engine priority: requests without a - /// platform priority carry the lowest value instead of leaving the body - /// fallback open, and the strict tier is pinned to the default. + /// Emit both priority headers on every inference request, so the engine + /// reads priority only from pylon and never from client-supplied values. pub(crate) fn apply_priority_headers( priority: Option, ceiling: u32, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index 24a3cc36f..103f99b31 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -669,7 +669,7 @@ pub(super) async fn forward_tunnel_request( // None for health requests, which skip header validation and are not // client inference traffic; nothing to trace or derive for them. - let upstream_context = lifecycle.as_ref().map(|lifecycle| UpstreamRequestContext { + let validated = lifecycle.as_ref().map(|lifecycle| ValidatedRequestContext { priority: lifecycle.required.priority, }); let response = match send_upstream_request( @@ -678,7 +678,7 @@ pub(super) async fn forward_tunnel_request( &path_and_query, &request_headers, body_bytes, - upstream_context, + validated, ) .await { @@ -715,7 +715,7 @@ pub(super) async fn forward_tunnel_request( /// `None` at the call site means a health request: unvalidated, untraced, /// and never carrying derived engine headers. #[derive(Clone, Copy)] -struct UpstreamRequestContext { +struct ValidatedRequestContext { /// Platform priority; `None` when the request carried no x-priority. priority: Option, } @@ -726,9 +726,9 @@ async fn send_upstream_request( path_and_query: &str, request_headers: &HeaderMap, body_bytes: Vec, - context: Option, + validated: Option, ) -> Result { - let span = if context.is_some() { + let span = if validated.is_some() { let span = tracing::info_span!( "pylon_upstream_http_request", otel_parent = field::Empty, @@ -752,18 +752,18 @@ async fn send_upstream_request( for (name, value) in request_headers { if should_forward_header(name, &app.retry) { upstream_headers.append(name, value.clone()); - } else if backend::dynamo::is_engine_priority_header(name) { + } else if backend::dynamo::is_engine_request_header(name) { // Values are client-controlled; log the name only. - tracing::debug!(header = %name, "stripped inbound engine priority header"); + tracing::debug!(header = %name, "stripped inbound engine request header"); } } - if let Some(context) = context { - if let Some(priority) = context.priority { + if let Some(validated) = validated { + if let Some(priority) = validated.priority { span.record("priority", priority); } if app.upstream_backend == UpstreamBackend::Dynamo { let dynamo_priority = backend::dynamo::apply_priority_headers( - context.priority, + validated.priority, app.priority_ceiling, &mut upstream_headers, ); @@ -1101,7 +1101,7 @@ pub(super) fn join_base_path(base: &str, path_and_query: &str) -> Result bool { !is_tunnel_control_header(name, retry) - && !backend::dynamo::is_engine_priority_header(name) + && !backend::dynamo::is_engine_request_header(name) && !matches!( name.as_str(), "host" | "x-method" | "x-path" | HEADER_STARGATE_EXPECTED_QUEUE_MS diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index 9796f8003..3ca92cf59 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -528,8 +528,7 @@ fn pylon_dynamo_priority_headers_are_always_emitted() { ); assert_eq!(headers["x-dynamo-request-strict-priority"], "0"); - // Absent platform priority pins both headers to the lowest values so the - // engine never falls back to client-controlled body hints. + // Absent platform priority pins both headers to the lowest values. let mut headers = HeaderMap::new(); let emitted = dynamo::apply_priority_headers(None, DEFAULT_PRIORITY_CEILING, &mut headers); assert_eq!(emitted, 0); @@ -1755,8 +1754,7 @@ async fn quic_tunnel_emits_lowest_dynamo_priority_without_x_priority() { .send(headers, br#"{"messages":[],"stream":true}"#) .await; - // Unconfigured requests carry the lowest priority instead of no header, - // so the engine never falls back to client-controlled body values. + // Unconfigured requests carry the lowest priority instead of no header. let response_headers = tunnel.response_head(StatusCode::OK).await; assert_eq!(response_headers["x-echo-dynamo-priority"], "0"); assert_eq!(response_headers["x-echo-dynamo-strict-priority"], "0"); diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs index 8f559b0b7..8aa0b6e0a 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs @@ -161,11 +161,12 @@ impl RequestObserver { generation: Option, runtime_state: PylonRuntimeState, ) -> Self { + let priority = required.queue_priority(); let RequiredTunnelHeaders { request_id, routing_key, model_id, - priority, + priority: _, input_tokens, accepted_at, } = required; @@ -175,7 +176,7 @@ impl RequestObserver { started_at: accepted_at, routing_key, model_id, - priority: priority.unwrap_or_default(), + priority, input_tokens, generation, embedding_items: None, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs index cc07559a2..8bbdc9e4f 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs @@ -62,13 +62,21 @@ pub(crate) struct RequiredTunnelHeaders { pub request_id: String, pub routing_key: Option, pub model_id: String, - /// `None` when the request carried no x-priority header. The distinction - /// from an explicit 0 matters to the engine priority derivation. + /// `None` when the request carried no x-priority header. Absent and an + /// explicit 0 are different values to the engine priority derivation. pub priority: Option, pub input_tokens: u64, pub(crate) accepted_at: Instant, } +impl RequiredTunnelHeaders { + /// Priority for queue accounting and observation, where unconfigured + /// counts as 0. The engine derivation reads `priority` directly instead. + pub(crate) fn queue_priority(&self) -> u32 { + self.priority.unwrap_or_default() + } +} + pub(crate) fn validate_required_tunnel_headers( request_headers: &HeaderMap, ) -> Result { diff --git a/src/libraries/rust/stargate/docs/api-gateway-contract.md b/src/libraries/rust/stargate/docs/api-gateway-contract.md index cc2afc3df..20b512155 100644 --- a/src/libraries/rust/stargate/docs/api-gateway-contract.md +++ b/src/libraries/rust/stargate/docs/api-gateway-contract.md @@ -156,8 +156,8 @@ Internal headers: derived from `x-priority` as `max(0, ceiling - x)` with a configurable ceiling (`--pylon-priority-ceiling`, default 3600), requests without `x-priority` carry the lowest value `0`, and the strict tier is always `0`. - Always emitting means the engine never falls back to client-controlled - request-body priority hints. + Always emitting means the engine reads priority only from pylon, never from + client-supplied values. Body rules: From 62bebdfc2ffe7f3e29d6e838333b5b748c9f1d14 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 21:37:51 -0700 Subject: [PATCH 07/17] refactor(pylon): pass health_request and priority to the upstream send path Replace the single-field ValidatedRequestContext with two plain parameters, matching the reviewer's suggestion to key the derive on health_request directly. Signed-off-by: along --- .../pylon-lib/src/quic_http_tunnel/core.rs | 35 ++++++++----------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index 103f99b31..d366bab07 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -667,18 +667,17 @@ pub(super) async fn forward_tunnel_request( } } - // None for health requests, which skip header validation and are not - // client inference traffic; nothing to trace or derive for them. - let validated = lifecycle.as_ref().map(|lifecycle| ValidatedRequestContext { - priority: lifecycle.required.priority, - }); + let priority = lifecycle + .as_ref() + .and_then(|lifecycle| lifecycle.required.priority); let response = match send_upstream_request( app, method, &path_and_query, &request_headers, body_bytes, - validated, + health_request, + priority, ) .await { @@ -711,24 +710,20 @@ pub(super) async fn forward_tunnel_request( Ok(()) } -/// Fields of the validated tunnel headers the upstream send path needs. -/// `None` at the call site means a health request: unvalidated, untraced, -/// and never carrying derived engine headers. -#[derive(Clone, Copy)] -struct ValidatedRequestContext { - /// Platform priority; `None` when the request carried no x-priority. - priority: Option, -} - +/// `health_request` requests skip header validation and are not client +/// inference traffic: they get no span, no trace context, and no derived +/// engine headers. `priority` is the validated x-priority value, `None` +/// when the header was absent (or the request is a health request). async fn send_upstream_request( app: &TunnelServerApp, method: Method, path_and_query: &str, request_headers: &HeaderMap, body_bytes: Vec, - validated: Option, + health_request: bool, + priority: Option, ) -> Result { - let span = if validated.is_some() { + let span = if !health_request { let span = tracing::info_span!( "pylon_upstream_http_request", otel_parent = field::Empty, @@ -757,13 +752,13 @@ async fn send_upstream_request( tracing::debug!(header = %name, "stripped inbound engine request header"); } } - if let Some(validated) = validated { - if let Some(priority) = validated.priority { + if !health_request { + if let Some(priority) = priority { span.record("priority", priority); } if app.upstream_backend == UpstreamBackend::Dynamo { let dynamo_priority = backend::dynamo::apply_priority_headers( - validated.priority, + priority, app.priority_ceiling, &mut upstream_headers, ); From 55e08cb87eea442dd948dbffbe3e72593d359d6a Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 21:55:18 -0700 Subject: [PATCH 08/17] refactor(pylon): scope the engine header strip to an explicit priority denylist Replace the x-dynamo-request- prefix match with a denylist of exactly the two priority headers pylon derives. Other engine headers pass through unchanged for now; widening the denylist is tracked separately. The filter test pins both the stripped names and the pass-through behavior. Signed-off-by: along --- .../pylon-lib/src/quic_http_tunnel/backend.rs | 13 +++++++----- .../pylon-lib/src/quic_http_tunnel/core.rs | 6 +++--- .../pylon-lib/src/quic_http_tunnel/tests.rs | 10 +++++++-- .../stargate/docs/api-gateway-contract.md | 21 ++++++++++--------- 4 files changed, 30 insertions(+), 20 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index 152367b69..a67f0045d 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -74,12 +74,15 @@ pub(crate) mod dynamo { /// shared tunnel contract because only pylon speaks them. pub(crate) const HEADER_REQUEST_PRIORITY: &str = "x-dynamo-request-priority"; pub(crate) const HEADER_REQUEST_STRICT_PRIORITY: &str = "x-dynamo-request-strict-priority"; - const REQUEST_HEADER_PREFIX: &str = "x-dynamo-request-"; - /// Pylon is the only writer of headers under this prefix, so inbound - /// values are stripped in every backend mode. - pub(crate) fn is_engine_request_header(name: &HeaderName) -> bool { - name.as_str().starts_with(REQUEST_HEADER_PREFIX) + /// Denylist of engine headers pylon owns: inbound values are stripped in + /// every backend mode so pylon stays their only writer. Scoped to the + /// priority headers for now; other engine headers are tracked separately. + const STRIPPED_REQUEST_HEADERS: [&str; 2] = + [HEADER_REQUEST_PRIORITY, HEADER_REQUEST_STRICT_PRIORITY]; + + pub(crate) fn is_stripped_engine_header(name: &HeaderName) -> bool { + STRIPPED_REQUEST_HEADERS.contains(&name.as_str()) } /// Map the platform rank (lower wins, absent = unconfigured) to the diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index d366bab07..bb20b0af9 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -747,9 +747,9 @@ async fn send_upstream_request( for (name, value) in request_headers { if should_forward_header(name, &app.retry) { upstream_headers.append(name, value.clone()); - } else if backend::dynamo::is_engine_request_header(name) { + } else if backend::dynamo::is_stripped_engine_header(name) { // Values are client-controlled; log the name only. - tracing::debug!(header = %name, "stripped inbound engine request header"); + tracing::debug!(header = %name, "stripped inbound engine priority header"); } } if !health_request { @@ -1096,7 +1096,7 @@ pub(super) fn join_base_path(base: &str, path_and_query: &str) -> Result bool { !is_tunnel_control_header(name, retry) - && !backend::dynamo::is_engine_request_header(name) + && !backend::dynamo::is_stripped_engine_header(name) && !matches!( name.as_str(), "host" | "x-method" | "x-path" | HEADER_STARGATE_EXPECTED_QUEUE_MS diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index 3ca92cf59..e1271eacb 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -480,7 +480,6 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() "X-Stargate-Expected-Queue-Ms", "X-Dynamo-Request-Priority", "X-Dynamo-Request-Strict-Priority", - "X-Dynamo-Request-Anything", ] .into_iter() .chain(RETRY_CONTROL_REQUEST_HEADERS) @@ -490,7 +489,14 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() &retry )); } - for name in [b"X-Request-Id".as_slice(), b"X-Priority", b"X-Dynamo-Nvext"] { + // The strip denylist is scoped to the priority headers; other engine + // headers pass through unchanged for now. + for name in [ + b"X-Request-Id".as_slice(), + b"X-Priority", + b"X-Dynamo-Nvext", + b"X-Dynamo-Request-Anything", + ] { assert!(should_forward_header( &HeaderName::from_bytes(name)?, &retry diff --git a/src/libraries/rust/stargate/docs/api-gateway-contract.md b/src/libraries/rust/stargate/docs/api-gateway-contract.md index 20b512155..dffbddb53 100644 --- a/src/libraries/rust/stargate/docs/api-gateway-contract.md +++ b/src/libraries/rust/stargate/docs/api-gateway-contract.md @@ -148,16 +148,17 @@ Internal headers: - `x-stargate-expected-queue-ms`: Stargate-to-pylon only. Stargate strips caller values; pylon strips it before upstream forwarding. -- `x-dynamo-request-*` (notably `x-dynamo-request-priority` and - `x-dynamo-request-strict-priority`): pylon-to-engine only. Pylon strips - inbound values in every backend mode, so pylon is the only writer of these - headers. When pylon runs with `--pylon-upstream-backend dynamo` (the - default), it emits both headers on every inference request: the priority is - derived from `x-priority` as `max(0, ceiling - x)` with a configurable - ceiling (`--pylon-priority-ceiling`, default 3600), requests without - `x-priority` carry the lowest value `0`, and the strict tier is always `0`. - Always emitting means the engine reads priority only from pylon, never from - client-supplied values. +- `x-dynamo-request-priority` and `x-dynamo-request-strict-priority`: + pylon-to-engine only. Pylon strips inbound values in every backend mode, so + pylon is the only writer of these two headers. When pylon runs with + `--pylon-upstream-backend dynamo` (the default), it emits both headers on + every inference request: the priority is derived from `x-priority` as + `max(0, ceiling - x)` with a configurable ceiling + (`--pylon-priority-ceiling`, default 3600), requests without `x-priority` + carry the lowest value `0`, and the strict tier is always `0`. Always + emitting means the engine reads priority only from pylon, never from + client-supplied values. Other engine headers pass through unchanged; the + strip denylist is scoped to the priority headers. Body rules: From fe05d7ce5fe845472a1be636be3e7dc9e3f41349 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 22:31:27 -0700 Subject: [PATCH 09/17] refactor(pylon): drop the strip debug log and inline the backend display The debug log duplicated the strip predicate in the forwarding loop with no consumer; spoof-attempt visibility can come back as a counter metric if someone needs it. as_str had Display as its only caller. Signed-off-by: along --- .../pylon-lib/src/quic_http_tunnel/backend.rs | 14 ++++---------- .../crates/pylon-lib/src/quic_http_tunnel/core.rs | 3 --- 2 files changed, 4 insertions(+), 13 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index a67f0045d..c4437cc0a 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -34,18 +34,12 @@ pub enum UpstreamBackend { Dynamo, } -impl UpstreamBackend { - pub fn as_str(&self) -> &'static str { - match self { - Self::Passthrough => "passthrough", - Self::Dynamo => "dynamo", - } - } -} - impl fmt::Display for UpstreamBackend { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(self.as_str()) + f.write_str(match self { + Self::Passthrough => "passthrough", + Self::Dynamo => "dynamo", + }) } } diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index bb20b0af9..40e5150c7 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -747,9 +747,6 @@ async fn send_upstream_request( for (name, value) in request_headers { if should_forward_header(name, &app.retry) { upstream_headers.append(name, value.clone()); - } else if backend::dynamo::is_stripped_engine_header(name) { - // Values are client-controlled; log the name only. - tracing::debug!(header = %name, "stripped inbound engine priority header"); } } if !health_request { From 0ea0c0db7e610a6bc198fb16e848f0523eda785c Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 23:12:53 -0700 Subject: [PATCH 10/17] docs(pylon): x-priority absence is not the same as rank 0 The optional-header table said x-priority defaults to 0, which invites a gateway to synthesize 0 for unconfigured requests. Absence maps to the lowest engine priority while rank 0 maps to the highest, so the gateway must omit the header when no priority resolves. Signed-off-by: along --- src/libraries/rust/stargate/docs/api-gateway-contract.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/libraries/rust/stargate/docs/api-gateway-contract.md b/src/libraries/rust/stargate/docs/api-gateway-contract.md index dffbddb53..9a70673ce 100644 --- a/src/libraries/rust/stargate/docs/api-gateway-contract.md +++ b/src/libraries/rust/stargate/docs/api-gateway-contract.md @@ -136,7 +136,7 @@ Optional trusted headers: | `x-routing-key` | Authenticated routing scope. Omit for unscoped. | | `x-routing-method` | Request-scoped load-balancer override, only for methods allowed by Stargate config. | | `x-cache-affinity-key` | Opaque cache/prefix identity. Required by some LB configs. | -| `x-priority` | Unsigned priority, default `0`. | +| `x-priority` | Unsigned priority rank; lower is more urgent and `0` is the most urgent. Omit when no priority is resolved. Never synthesize `0` for unconfigured requests: absence maps to the lowest engine priority, while `0` maps to the highest. Stargate treats an absent header as `0` for its own queue accounting only. | | `x-request-slo-ms` | Per-request LB latency hint. | | `x-max-wait-ms` | Wait budget for temporarily infeasible candidates. | | `x-stargate-max-wait-ms` | Stargate internal retry budget. | From b24d9aee2b68dd010e8b9278399526bc3c3a1e5c Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 23:17:32 -0700 Subject: [PATCH 11/17] docs(pylon): document absent-vs-0 x-priority semantics in the internal headers section Keep the optional-header table row short and put the explanation where the derivation is documented: absence and rank 0 are opposite ends of the range, so the gateway must not synthesize 0 for unconfigured requests. Signed-off-by: along --- src/libraries/rust/stargate/docs/api-gateway-contract.md | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/libraries/rust/stargate/docs/api-gateway-contract.md b/src/libraries/rust/stargate/docs/api-gateway-contract.md index 9a70673ce..5c7323f6d 100644 --- a/src/libraries/rust/stargate/docs/api-gateway-contract.md +++ b/src/libraries/rust/stargate/docs/api-gateway-contract.md @@ -136,7 +136,7 @@ Optional trusted headers: | `x-routing-key` | Authenticated routing scope. Omit for unscoped. | | `x-routing-method` | Request-scoped load-balancer override, only for methods allowed by Stargate config. | | `x-cache-affinity-key` | Opaque cache/prefix identity. Required by some LB configs. | -| `x-priority` | Unsigned priority rank; lower is more urgent and `0` is the most urgent. Omit when no priority is resolved. Never synthesize `0` for unconfigured requests: absence maps to the lowest engine priority, while `0` maps to the highest. Stargate treats an absent header as `0` for its own queue accounting only. | +| `x-priority` | Unsigned priority rank; lower is more urgent. Omit when no priority is resolved. | | `x-request-slo-ms` | Per-request LB latency hint. | | `x-max-wait-ms` | Wait budget for temporarily infeasible candidates. | | `x-stargate-max-wait-ms` | Stargate internal retry budget. | @@ -160,6 +160,12 @@ Internal headers: client-supplied values. Other engine headers pass through unchanged; the strip denylist is scoped to the priority headers. + An absent `x-priority` and `x-priority: 0` are opposite ends of the range: + absence maps to the lowest engine priority, while rank `0` maps to the + highest. The gateway must not synthesize `x-priority: 0` for unconfigured + requests. Stargate treats an absent header as `0` for its own queue + accounting only; that default never reaches the engine. + Body rules: - Stargate treats bodies as opaque bytes. From f1c6bb3efce07e17a56d1524f4acccef916e5b17 Mon Sep 17 00:00:00 2001 From: along Date: Mon, 10 Aug 2026 23:48:49 -0700 Subject: [PATCH 12/17] docs(pylon): make the priority ceiling flag help backend-neutral SECONDS is the Dynamo interpretation of the derived value; the flag itself is a platform rank band, so name the placeholder RANK and note the Dynamo unit in the help text. Signed-off-by: along --- src/libraries/rust/stargate/crates/pylon/src/main.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon/src/main.rs b/src/libraries/rust/stargate/crates/pylon/src/main.rs index 29718ea57..07d6a61a4 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/main.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/main.rs @@ -216,13 +216,14 @@ struct Args { value_name = "BACKEND" )] pylon_upstream_backend: UpstreamBackend, - /// Seconds of engine scheduling head start for the most urgent platform - /// priority (x-priority: 0); lower x-priority ranks get proportionally less + /// Priority band ceiling: x-priority rank 0 maps to this engine value and + /// ranks at or beyond it map to the lowest. Dynamo reads the derived + /// value as seconds of queue head start. #[arg( long, default_value_t = pylon_lib::DEFAULT_PRIORITY_CEILING, env = "PYLON_PRIORITY_CEILING", - value_name = "SECONDS" + value_name = "RANK" )] pylon_priority_ceiling: u32, /// Collect post-stream output quality metrics (gibberish checks) From ab21feaea99982e3f053b18539762916e965260a Mon Sep 17 00:00:00 2001 From: along Date: Tue, 11 Aug 2026 00:11:17 -0700 Subject: [PATCH 13/17] test(pylon): pin health-request and zero-ceiling priority behavior Health requests must carry no derived engine headers, and a ceiling of 0 collapses every rank to the lowest value. Also tighten two doc comments. Signed-off-by: along --- .../pylon-lib/src/quic_http_tunnel/backend.rs | 2 +- .../pylon-lib/src/quic_http_tunnel/core.rs | 8 ++--- .../pylon-lib/src/quic_http_tunnel/tests.rs | 35 ++++++++++++++++++- 3 files changed, 39 insertions(+), 6 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index c4437cc0a..995625433 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -26,7 +26,7 @@ use std::str::FromStr; /// add a variant and a submodule here, never a new CLI flag. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum UpstreamBackend { - /// Forward requests unchanged; derive nothing. Inbound engine request + /// Forward requests unchanged; derive nothing. Inbound engine priority /// headers are still stripped. Passthrough, /// Derive the engine priority headers from `x-priority`. diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index 40e5150c7..4ec3de754 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -710,10 +710,10 @@ pub(super) async fn forward_tunnel_request( Ok(()) } -/// `health_request` requests skip header validation and are not client -/// inference traffic: they get no span, no trace context, and no derived -/// engine headers. `priority` is the validated x-priority value, `None` -/// when the header was absent (or the request is a health request). +/// Health requests skip header validation and are not client inference +/// traffic: no span, no trace context, no derived engine headers. +/// `priority` is the validated x-priority value; `None` when the header was +/// absent or the request is a health request. async fn send_upstream_request( app: &TunnelServerApp, method: Method, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index e1271eacb..073079647 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -521,6 +521,9 @@ fn pylon_dynamo_request_priority_inverts_within_bounded_ceiling() { // A ceiling beyond i32 is clamped so the emitted value stays a valid i32. assert_eq!(dynamo::request_priority(Some(0), u32::MAX), i32::MAX); assert_eq!(dynamo::request_priority(None, u32::MAX), 0); + // A ceiling of 0 collapses every rank to the lowest value. + assert_eq!(dynamo::request_priority(Some(0), 0), 0); + assert_eq!(dynamo::request_priority(None, 0), 0); } #[test] @@ -1685,7 +1688,20 @@ async fn quic_tunnel_forwards_to_http_backend() { /// Echoes the Dynamo priority headers the backend received, so the tunnel /// tests assert on what actually crossed the pylon-to-engine hop. fn dynamo_priority_echo_router() -> Router { - Router::new().route( + Router::new() + .route( + "/health", + axum::routing::get(|req: Request| async move { + let dynamo_priority = req + .headers() + .get("x-dynamo-request-priority") + .and_then(|value| value.to_str().ok()) + .unwrap_or("absent") + .to_string(); + ([("x-echo-dynamo-priority", dynamo_priority)], "ok") + }), + ) + .route( "/v1/chat/completions", post(|req: Request| async move { let echo_header = |name: &str| { @@ -1768,6 +1784,23 @@ async fn quic_tunnel_emits_lowest_dynamo_priority_without_x_priority() { tunnel.shutdown().await; } +#[tokio::test] +async fn quic_tunnel_health_requests_carry_no_derived_priority() { + let (config, _metrics) = metered_test_tunnel_config_for(dynamo_priority_echo_router()).await; + let mut tunnel = RawTunnelTest::start(config).await; + + // Health requests skip validation, so no required tunnel headers. + let mut headers = HeaderMap::new(); + headers.insert("x-method", "GET".parse().unwrap()); + headers.insert("x-path", "/health".parse().unwrap()); + tunnel.send(headers, b"").await; + + let response_headers = tunnel.response_head(StatusCode::OK).await; + assert_eq!(response_headers["x-echo-dynamo-priority"], "absent"); + + tunnel.shutdown().await; +} + #[tokio::test] async fn quic_tunnel_passthrough_backend_strips_but_derives_nothing() { let (mut config, _metrics) = From a99ad4ed2a620a0a9e76a7a70b2ff62223d7d4b6 Mon Sep 17 00:00:00 2001 From: along Date: Tue, 11 Aug 2026 08:27:04 -0700 Subject: [PATCH 14/17] docs(pylon): state each priority rule once and reference it elsewhere The strip-in-every-mode rule lives on the denylist and the Passthrough variant; the ceiling semantics live on the mapping function. Config fields, flag help, and a test comment now reference instead of restate. Signed-off-by: along --- .../crates/pylon-lib/src/quic_http_tunnel/backend.rs | 3 +-- .../stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs | 6 ++---- .../rust/stargate/crates/pylon-lib/src/request_observer.rs | 2 -- src/libraries/rust/stargate/crates/pylon/src/main.rs | 3 +-- 4 files changed, 4 insertions(+), 10 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index 995625433..211ad50e1 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -57,8 +57,7 @@ impl FromStr for UpstreamBackend { } } -/// Default seconds of scheduling head start for the most urgent platform -/// priority (`x-priority: 0`); see [`dynamo::request_priority`]. +/// Default priority band ceiling; see [`dynamo::request_priority`]. pub const DEFAULT_PRIORITY_CEILING: u32 = 3600; pub(crate) mod dynamo { diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index 4ec3de754..e05235a36 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -106,11 +106,9 @@ pub struct TunnelForwardingConfig { pub request_quality_monitor: RequestQualityMonitorConfig, pub retry: PylonRetryConfig, pub queue_mismatch_retry: PylonQueueMismatchRetryConfig, - /// Engine dialect spoken to the local upstream. Inbound engine priority - /// headers are stripped in every mode; only derivation is per-backend. + /// Engine dialect spoken to the local upstream; see [`UpstreamBackend`]. pub upstream_backend: UpstreamBackend, - /// Seconds of scheduling head start for the most urgent platform - /// priority; see [`backend::dynamo::request_priority`]. + /// Priority band ceiling; see [`backend::dynamo::request_priority`]. pub priority_ceiling: u32, pub metrics: Option>, #[cfg(test)] diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs index 8aa0b6e0a..84c1192b3 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer.rs @@ -613,8 +613,6 @@ mod tests { fn validate_required_tunnel_headers_keeps_missing_priority_absent() { let required = validate_required_tunnel_headers(&request_headers("req-1", 42)).unwrap(); - // Absent stays absent rather than defaulting to 0: the engine - // priority derivation treats unconfigured and rank 0 differently. assert_eq!(required.priority, None); } diff --git a/src/libraries/rust/stargate/crates/pylon/src/main.rs b/src/libraries/rust/stargate/crates/pylon/src/main.rs index 07d6a61a4..609a15a58 100644 --- a/src/libraries/rust/stargate/crates/pylon/src/main.rs +++ b/src/libraries/rust/stargate/crates/pylon/src/main.rs @@ -207,8 +207,7 @@ struct Args { #[arg(long, env = "PYLON_QUEUE_MISMATCH_RETRY_AFTER_MS", value_name = "MS")] pylon_queue_mismatch_retry_after_ms: Option, /// Engine dialect spoken to the local upstream: "dynamo" derives the - /// engine priority headers from x-priority, "passthrough" derives nothing. - /// Inbound engine priority headers are stripped in every mode. + /// engine priority headers from x-priority, "passthrough" derives nothing #[arg( long, default_value = "dynamo", From 8c403e12e5ed545d7cb5b579ecc25d485579ed8b Mon Sep 17 00:00:00 2001 From: along Date: Tue, 11 Aug 2026 10:41:21 -0700 Subject: [PATCH 15/17] test(pylon): restore original pass-through assertion in filter test The pass-through loop added X-Priority, X-Dynamo-Nvext, and X-Dynamo-Request-Anything assertions that pin no behavior this change introduces; keep only the pre-existing X-Request-Id check. Signed-off-by: along --- .../pylon-lib/src/quic_http_tunnel/tests.rs | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index 073079647..735a64fc1 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -489,19 +489,10 @@ fn pylon_request_header_filter_strips_tunnel_headers_case_insensitively() &retry )); } - // The strip denylist is scoped to the priority headers; other engine - // headers pass through unchanged for now. - for name in [ - b"X-Request-Id".as_slice(), - b"X-Priority", - b"X-Dynamo-Nvext", - b"X-Dynamo-Request-Anything", - ] { - assert!(should_forward_header( - &HeaderName::from_bytes(name)?, - &retry - )); - } + assert!(should_forward_header( + &HeaderName::from_bytes(b"X-Request-Id")?, + &retry + )); Ok(()) } From 070007d6dd4bb0b074b7853c8b6db8875ee2af86 Mon Sep 17 00:00:00 2001 From: along Date: Tue, 11 Aug 2026 10:56:15 -0700 Subject: [PATCH 16/17] test(pylon): pin spoofed engine header strip on the health path The health tunnel test now sends a spoofed x-dynamo-request-priority; the existing absent assertion proves it is stripped and not replaced. Signed-off-by: along --- .../stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs index 735a64fc1..ae8ef3dab 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/tests.rs @@ -1784,6 +1784,8 @@ async fn quic_tunnel_health_requests_carry_no_derived_priority() { let mut headers = HeaderMap::new(); headers.insert("x-method", "GET".parse().unwrap()); headers.insert("x-path", "/health".parse().unwrap()); + // A spoofed engine header is stripped on the health path too. + headers.insert("x-dynamo-request-priority", "42".parse().unwrap()); tunnel.send(headers, b"").await; let response_headers = tunnel.response_head(StatusCode::OK).await; From 6cfc5d1325538106d539352c1a640738a58788e0 Mon Sep 17 00:00:00 2001 From: along Date: Tue, 11 Aug 2026 10:57:12 -0700 Subject: [PATCH 17/17] fix: trim comments --- .../stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs | 3 +-- .../stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs | 4 ---- .../stargate/crates/pylon-lib/src/request_observer/headers.rs | 2 -- 3 files changed, 1 insertion(+), 8 deletions(-) diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs index 211ad50e1..c797388e4 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs @@ -88,8 +88,7 @@ pub(crate) mod dynamo { (ceiling - rank) as i32 } - /// Emit both priority headers on every inference request, so the engine - /// reads priority only from pylon and never from client-supplied values. + /// Emit both priority headers on every inference request. pub(crate) fn apply_priority_headers( priority: Option, ceiling: u32, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs index e05235a36..22916aead 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs @@ -708,10 +708,6 @@ pub(super) async fn forward_tunnel_request( Ok(()) } -/// Health requests skip header validation and are not client inference -/// traffic: no span, no trace context, no derived engine headers. -/// `priority` is the validated x-priority value; `None` when the header was -/// absent or the request is a health request. async fn send_upstream_request( app: &TunnelServerApp, method: Method, diff --git a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs index 8bbdc9e4f..2b11e8f4c 100644 --- a/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs +++ b/src/libraries/rust/stargate/crates/pylon-lib/src/request_observer/headers.rs @@ -62,8 +62,6 @@ pub(crate) struct RequiredTunnelHeaders { pub request_id: String, pub routing_key: Option, pub model_id: String, - /// `None` when the request carried no x-priority header. Absent and an - /// explicit 0 are different values to the engine priority derivation. pub priority: Option, pub input_tokens: u64, pub(crate) accepted_at: Instant,