fix(stargate): bound proxy retries and relay retention

AGENTS.md

@@ -17,6 +17,8 @@ Useful pointers:
 - `BAZEL.md` for the contributor-facing Bazel build path
 - `docs/AGENTS.md` for in-repo user and developer documentation
 - `tools/AGENTS.md` for repo tooling
+- `src/libraries/rust/stargate/AGENTS.md` for Stargate deployment invariants,
+  including router transport assumptions to use during implementation and review
 - `imports.yaml` for subtree ownership and commit pins
 - `.cursor/skills/documentation-style/SKILL.md` for docs style
 - `.cursor/skills/` for root dev-skill symlink fanout

src/libraries/rust/stargate/AGENTS.md

@@ -0,0 +1,38 @@
+# Stargate agent guide
+
+This guidance covers the Rust workspace, including `stargate-k8s-router` and
+the shared `stargate-forwarding` crate.
+
+## Assumed deployment invariants
+
+- `stargate-k8s-router` supports `raw-quic` and `webtransport` tunnel traffic.
+  Plain `http3` tunnel traffic uses an L4 path and does not pass through this router.
+- Raw QUIC request streams are bidirectional. During drain, prioritizing
+  bidirectional acceptance over unidirectional acceptance is intentional.
+- WebTransport uses a separate session bridge. Do not infer a plain HTTP/3
+  stream fairness requirement for the raw QUIC relay from that implementation.
+
+For router transport changes, inspect `crates/stargate-k8s-router/src/quic.rs`
+and `crates/stargate-forwarding/src/lib.rs`. The
+[transport guide](docs/tunnel-transports.md) describes available protocol
+implementations and their routing paths.
+
+## Build and verification
+
+Run commands from this directory. For router and relay Rust changes:
+
+```sh
+cargo build --locked -p stargate-k8s-router
+cargo test --locked -p stargate-k8s-router -p stargate-forwarding -- --test-threads=1
+cargo clippy --locked -p stargate-k8s-router -p stargate-forwarding --all-targets -- -D warnings
+cargo fmt --all -- --check
+```
+
+For other Rust changes, select the affected workspace packages with `-p`.
+For documentation changes, check whitespace and link targets.
+
+## Code style
+
+Use Rust 2024 and the workspace formatter and Clippy rules. Match existing
+crate structure and use `tracing` for structured logs. Keep deployment
+assumptions explicit when reviewing transport behavior.

src/libraries/rust/stargate/CLAUDE.md

@@ -0,0 +1 @@
+@AGENTS.md

src/libraries/rust/stargate/crates/stargate-forwarding/src/lib.rs

@@ -300,6 +300,11 @@ async fn relay_direction(acceptor: quinn::Connection, initiator: quinn::Connecti
     let mut tasks = tokio::task::JoinSet::new();
     loop {
         tokio::select! {
+            result = tasks.join_next(), if !tasks.is_empty() => {
+                if let Some(Err(error)) = result {
+                    warn!(%error, "stream relay task failed");
+                }
+            }
             bi = acceptor.accept_bi() => {
                 spawn_stream_relay!(tasks, bi, initiator, relay_bi_stream,
                     "accept_bi failed in relay", "bi-stream relay error");
@@ -323,6 +328,11 @@ async fn relay_direction_until_shutdown(
         tokio::select! {
             biased;
             _ = drain.changed() => break,
+            result = tasks.join_next(), if !tasks.is_empty() => {
+                if let Some(Err(error)) = result {
+                    warn!(%error, "draining stream relay task failed");
+                }
+            }
             bi = acceptor.accept_bi() => {
                 spawn_stream_relay!(tasks, bi, initiator, relay_bi_stream_until_delivered,
                     "accept_bi failed in draining relay", "draining bi-stream relay error");

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

@@ -134,13 +134,20 @@ impl ProxyRequestRun<'_> {
             selected.expected_queue_ms,
         );
         let upstream_start = Instant::now();
+        let mut request_body_started = false;
         let upstream = proxy_via_quic_streaming(
             self.app,
             &chosen.registration,
             self.request.method.clone(),
             &self.request.path_and_query,
             attempt_headers,
-            || self.request.replay_body.body_for_attempt(),
+            || {
+                let body = self.request.replay_body.body_for_attempt()?;
+                // After handing the body to the tunnel, a transport failure
+                // cannot establish whether the backend applied this POST.
+                request_body_started = true;
+                Ok(body)
+            },
         )
         .instrument(upstream_span.clone())
         .await;
@@ -154,6 +161,7 @@ impl ProxyRequestRun<'_> {
                     &self.app.retry,
                     retry_budget_has_remaining(self.request.retry_deadline),
                     self.attempt_counters.connect_retries,
+                    request_body_started,
                     self.request.replay_body.replay_readiness(),
                 ) {
                     RetryDecision::Final(disposition) => {
@@ -361,6 +369,10 @@ fn finish_attempt(
     }
     let upstream = match disposition {
         FinalRetryDisposition::PassThrough | FinalRetryDisposition::ReplayIncomplete(_) => upstream,
+        FinalRetryDisposition::AmbiguousDelivery => {
+            metrics.proxy_ambiguous_delivery_total().inc();
+            upstream
+        }
         FinalRetryDisposition::Exhausted(retry_reason) => {
             metrics
                 .proxy_retry_exhausted_total(run.routing_key(), run.model_id(), &retry_reason)

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

@@ -66,6 +66,7 @@ pub(super) enum RetryDecision<T> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) enum FinalRetryDisposition {
     PassThrough,
+    AmbiguousDelivery,
     Exhausted(String),
     ReplayIncomplete(String),
     PayloadTooLarge(Option<String>),
@@ -75,6 +76,7 @@ impl FinalRetryDisposition {
     pub(super) fn label(&self) -> &'static str {
         match self {
             Self::PassThrough => "pass_through",
+            Self::AmbiguousDelivery => "ambiguous_delivery",
             Self::Exhausted(_) => "retry_exhausted",
             Self::ReplayIncomplete(_) => "replay_incomplete",
             Self::PayloadTooLarge(_) => "payload_too_large",
@@ -84,6 +86,7 @@ impl FinalRetryDisposition {
     pub(super) fn retry_reason(&self) -> Option<&str> {
         match self {
             Self::PassThrough => None,
+            Self::AmbiguousDelivery => Some("request_may_have_been_applied"),
             Self::Exhausted(reason) | Self::ReplayIncomplete(reason) => Some(reason),
             Self::PayloadTooLarge(reason) => reason.as_deref(),
         }
@@ -137,6 +140,7 @@ pub(super) fn decide_proxy_error_retry(
     retry: &ProxyRetryConfig,
     retry_budget_remaining: bool,
     connect_retries: u32,
+    request_body_started: bool,
     replay_readiness: ReplayReadiness,
 ) -> RetryDecision<()> {
     if !matches!(
@@ -145,6 +149,9 @@ pub(super) fn decide_proxy_error_retry(
     ) {
         return RetryDecision::Final(FinalRetryDisposition::PassThrough);
     }
+    if request_body_started && matches!(replay_readiness, ReplayReadiness::Ready) {
+        return RetryDecision::Final(FinalRetryDisposition::AmbiguousDelivery);
+    }
     if !retry_budget_remaining {
         return retry_exhausted("retry_budget_exhausted");
     }
@@ -278,6 +285,36 @@ mod tests {
         .collect()
     }
 
+    #[test]
+    fn completed_body_does_not_authorize_retry_after_submission() {
+        let retry = ProxyRetryConfig::default();
+        for status in [
+            StatusCode::BAD_GATEWAY,
+            StatusCode::GATEWAY_TIMEOUT,
+            StatusCode::SERVICE_UNAVAILABLE,
+        ] {
+            for (budget_remaining, connect_retries) in [
+                (true, 0),
+                (false, 0),
+                (true, retry.max_connect_retries),
+                (false, retry.max_connect_retries),
+            ] {
+                assert_eq!(
+                    decide_proxy_error_retry(
+                        status,
+                        &retry,
+                        budget_remaining,
+                        connect_retries,
+                        true,
+                        ReplayReadiness::Ready,
+                    ),
+                    RetryDecision::Final(FinalRetryDisposition::AmbiguousDelivery),
+                    "status={status}, budget_remaining={budget_remaining}, connect_retries={connect_retries}",
+                );
+            }
+        }
+    }
+
     #[test]
     fn capacity_rejection_requires_trusted_admission_metadata() {
         for reason in [
@@ -491,6 +528,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::Ready,
             ),
             RetryDecision::Retry(())
@@ -507,6 +545,7 @@ mod tests {
                 &retry,
                 false,
                 0,
+                false,
                 ReplayReadiness::Ready,
             ),
             RetryDecision::Final(FinalRetryDisposition::Exhausted(
@@ -519,6 +558,7 @@ mod tests {
                 &retry,
                 true,
                 retry.max_connect_retries,
+                false,
                 ReplayReadiness::PayloadTooLarge,
             ),
             RetryDecision::Final(FinalRetryDisposition::Exhausted(
@@ -531,6 +571,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::PayloadTooLarge,
             ),
             RetryDecision::Final(FinalRetryDisposition::PassThrough)
@@ -547,6 +588,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::Incomplete,
             ),
             RetryDecision::Final(FinalRetryDisposition::ReplayIncomplete(

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

@@ -179,7 +179,8 @@ pub(super) fn finalize_no_routing_choice(
         NoRoutingFinalization::NoCandidatesNotFound => {
             context
                 .metrics
-                .requests_total(rk_ref, model_id, "", "404")
+                // Unregistered request headers must not create metric series.
+                .requests_total(None, "", "", "404")
                 .inc();
             Ok(no_eligible_candidates_response())
         }
@@ -359,6 +360,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn unknown_targets_share_one_metric_series() {
+        let metrics = StargateMetrics::new().unwrap();
+        for index in 0..128 {
+            let target = RoutingTargetKey::new(
+                Some(format!("unknown-tenant-{index}")),
+                format!("unknown-model-{index}"),
+            );
+            let response = finalize_no_routing_choice(NoRoutingFinalizationContext {
+                metrics: &metrics,
+                target: &target,
+                finalization: NoRoutingFinalization::NoCandidatesNotFound,
+                failed_backend_count: 0,
+                failed_cluster_count: 0,
+                routing_retry_attempts: 0,
+                capacity_rejected: false,
+                routing_capacity_rejected: false,
+            })
+            .unwrap();
+            assert_eq!(response.status(), StatusCode::NOT_FOUND);
+        }
+        let family = metrics
+            .registry()
+            .gather()
+            .into_iter()
+            .find(|family| family.name() == "stargate_requests_total")
+            .unwrap();
+        assert_eq!(family.get_metric().len(), 1);
+        assert_eq!(family.get_metric()[0].get_counter().value(), 128.0);
+    }
+
     #[test]
     fn input_work_admission_rejects_overloaded_pool() {
         let mut candidate = cluster_candidate("cluster-a");

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

@@ -101,6 +101,7 @@ define_stargate_metrics! {
         routing_selections_total("routing_selections_total", "Total number of primary and ranked fallback cluster choices used for upstream attempts", ["routing_key", "model", "algorithm", "selection"]);
         routing_kv_free_token_fallback_selections_total("routing_kv_free_token_fallback_selections_total", "Total number of selected routes reached after a higher-ranked candidate was skipped by KV free-token eligibility", ["routing_key", "model", "algorithm"]);
         proxy_retry_exhausted_total("proxy_retry_exhausted_total", "Total number of proxy requests that exhausted retry options", ["routing_key", "model", "reason"]);
+        proxy_ambiguous_delivery_total("proxy_ambiguous_delivery_total", "Total number of proxy requests stopped because delivery may have occurred", []);
         admission_rejections_total("admission_rejections_total", "Total number of requests rejected by local admission control", ["routing_key", "model", "reason"]);
         quic_connection_evictions_total("quic_connection_evictions_total", "Total number of QUIC connection pool evictions", ["inference_server_id", "reason"]);
         quic_hot_path_reconnect_total("quic_hot_path_reconnect_total", "Total number of direct QUIC reconnects attempted on the proxy hot path", ["inference_server_id", "result"]);
@@ -130,7 +131,7 @@ macro_rules! metric_accessors {
         $(
             #[inline]
             pub fn $name(&self, $($arg: $arg_type),*) -> $return_type {
-                self.$name.with_label_values(&[$($label),*])
+                self.$name.with_label_values::<&str>(&[$($label),*])
             }
         )*
     };
@@ -143,6 +144,8 @@ impl StargateMetrics {
 
     pub fn new_with_prefix(prefix: &str) -> anyhow::Result<Arc<Self>> {
         let metrics = Arc::new(Self::register(prefix)?);
+        metrics.requests_total(None, "", "", "404").inc_by(0);
+        metrics.proxy_ambiguous_delivery_total().inc_by(0);
         for outcome in stargate_tls::TlsReloadOutcome::ALL {
             metrics
                 .tls_reloads_total
@@ -189,6 +192,7 @@ impl StargateMetrics {
         GenericCounter<AtomicU64>, routing_selections_total(routing_key: Option<&str>, model: &str, algorithm: &str, selection: &str) => [routing_key.unwrap_or(""), model, algorithm, selection];
         GenericCounter<AtomicU64>, routing_kv_free_token_fallback_selections_total(routing_key: Option<&str>, model: &str, algorithm: &str) => [routing_key.unwrap_or(""), model, algorithm];
         GenericCounter<AtomicU64>, proxy_retry_exhausted_total(routing_key: Option<&str>, model: &str, reason: &str) => [routing_key.unwrap_or(""), model, reason];
+        GenericCounter<AtomicU64>, proxy_ambiguous_delivery_total() => [];
         GenericCounter<AtomicU64>, admission_rejections_total(routing_key: Option<&str>, model: &str, reason: &str) => [routing_key.unwrap_or(""), model, reason];
         GenericCounter<AtomicU64>, quic_connection_evictions_total(inference_server_id: &str, reason: &str) => [inference_server_id, reason];
         GenericCounter<AtomicU64>, quic_hot_path_reconnect_total(inference_server_id: &str, result: &str) => [inference_server_id, result];
@@ -321,6 +325,8 @@ mod tests {
     #[test]
     fn default_metrics_prefix_keeps_stargate_metric_names() {
         let metrics = StargateMetrics::new().expect("metrics should initialize");
+        let initial = metrics.gather_text().expect("metrics should encode");
+        assert!(initial.contains("stargate_proxy_ambiguous_delivery_total 0\n"));
 
         metrics
             .requests_total(None, "model-a", "server-a", "200")