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,39 @@
+# 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` is deployed only for `raw-quic` tunnel traffic.
+  HTTP/3 tunnel traffic does not pass through this router.
+- Raw QUIC request streams are bidirectional. Prioritizing bidirectional
+  acceptance over unidirectional acceptance in the raw QUIC relay is intentional.
+- Optional HTTP/3 and WebTransport implementations, flags, and tests do not
+  change this deployment assumption. Do not infer an HTTP/3 stream fairness
+  requirement for the raw QUIC router from their presence.
+
+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,13 @@ async fn relay_direction(acceptor: quinn::Connection, initiator: quinn::Connecti
     let mut tasks = tokio::task::JoinSet::new();
     loop {
         tokio::select! {
+            biased;
+            result = tasks.join_next(), if !tasks.is_empty() => {
+                if let Some(Err(error)) = result {
+                    warn!(%error, "stream relay task failed");
+                }
+            }
+            // The raw-QUIC router's request streams are bidirectional; prefer them over uni streams.
             bi = acceptor.accept_bi() => {
                 spawn_stream_relay!(tasks, bi, initiator, relay_bi_stream,
                     "accept_bi failed in relay", "bi-stream relay error");
@@ -323,6 +330,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

@@ -132,13 +132,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;
@@ -152,6 +159,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) => {
@@ -356,7 +364,9 @@ fn finish_attempt(
         Span::current().record("proxy.retry_reason", retry_reason);
     }
     let upstream = match disposition {
-        FinalRetryDisposition::PassThrough | FinalRetryDisposition::ReplayIncomplete(_) => upstream,
+        FinalRetryDisposition::PassThrough
+        | FinalRetryDisposition::AmbiguousDelivery
+        | FinalRetryDisposition::ReplayIncomplete(_) => 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

@@ -64,6 +64,7 @@ pub(super) enum RetryDecision<T> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) enum FinalRetryDisposition {
     PassThrough,
+    AmbiguousDelivery,
     Exhausted(String),
     ReplayIncomplete(String),
     PayloadTooLarge(Option<String>),
@@ -73,6 +74,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",
@@ -82,6 +84,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(),
         }
@@ -135,6 +138,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!(
@@ -151,7 +155,8 @@ pub(super) fn decide_proxy_error_retry(
     }
 
     match replay_readiness {
-        ReplayReadiness::Ready => RetryDecision::Retry(()),
+        ReplayReadiness::Ready if !request_body_started => RetryDecision::Retry(()),
+        ReplayReadiness::Ready => RetryDecision::Final(FinalRetryDisposition::AmbiguousDelivery),
         ReplayReadiness::Incomplete => RetryDecision::Final(
             FinalRetryDisposition::ReplayIncomplete(RETRY_REASON_RETRYABLE_PROXY_ERROR.to_string()),
         ),
@@ -258,6 +263,21 @@ mod tests {
         .collect()
     }
 
+    #[test]
+    fn completed_body_does_not_authorize_retry_after_submission() {
+        assert_eq!(
+            decide_proxy_error_retry(
+                StatusCode::BAD_GATEWAY,
+                &ProxyRetryConfig::default(),
+                true,
+                0,
+                true,
+                ReplayReadiness::Ready,
+            ),
+            RetryDecision::Final(FinalRetryDisposition::AmbiguousDelivery)
+        );
+    }
+
     #[test]
     fn retry_requires_explicit_pylon_signal_by_default() {
         let retry = ProxyRetryConfig::default();
@@ -409,6 +429,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::Ready,
             ),
             RetryDecision::Retry(())
@@ -425,6 +446,7 @@ mod tests {
                 &retry,
                 false,
                 0,
+                false,
                 ReplayReadiness::Ready,
             ),
             RetryDecision::Final(FinalRetryDisposition::Exhausted(
@@ -437,6 +459,7 @@ mod tests {
                 &retry,
                 true,
                 retry.max_connect_retries,
+                false,
                 ReplayReadiness::PayloadTooLarge,
             ),
             RetryDecision::Final(FinalRetryDisposition::Exhausted(
@@ -449,6 +472,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::PayloadTooLarge,
             ),
             RetryDecision::Final(FinalRetryDisposition::PassThrough)
@@ -465,6 +489,7 @@ mod tests {
                 &retry,
                 true,
                 0,
+                false,
                 ReplayReadiness::Incomplete,
             ),
             RetryDecision::Final(FinalRetryDisposition::ReplayIncomplete(

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

@@ -164,7 +164,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())
         }
@@ -323,6 +324,35 @@ 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,
+            })
+            .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

@@ -143,6 +143,7 @@ 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);
         for outcome in stargate_tls::TlsReloadOutcome::ALL {
             metrics
                 .tls_reloads_total

src/libraries/rust/stargate/crates/stargate/tests/suite/proxy_contract.rs

@@ -2921,3 +2921,100 @@ async fn pulsar_missing_input_tokens_header_returns_400() {
         .assert_missing_header("req-no-input-tokens", ("x-cache-affinity-key", "prefix-a"))
         .await;
 }
+
+#[tokio::test]
+async fn submitted_post_is_not_replayed_after_header_timeout() {
+    init_crypto();
+    let model = "review-duplicate-post";
+    let hits = Arc::new(AtomicUsize::new(0));
+    let backend_hits = hits.clone();
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let backend_addr = listener.local_addr().unwrap();
+    let app = Router::new()
+        .route("/health", get(|| async { "ok" }))
+        .route("/v1/chat/completions", post(move |req: Request| {
+            let hits = backend_hits.clone();
+            async move {
+                axum::body::to_bytes(req.into_body(), 1024 * 1024).await.unwrap();
+                let prior = hits.fetch_add(1, Ordering::SeqCst);
+                if prior == 0 {
+                    tokio::time::sleep(Duration::from_millis(900)).await;
+                }
+                Response::builder()
+                    .header("content-type", "text/event-stream")
+                    .body(Body::from("data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\ndata: [DONE]\n\n"))
+                    .unwrap()
+            }
+        }));
+    let backend_task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+    let (grpc_addr, grpc_listener) = bind_ephemeral();
+    let (model_addr, model_listener) = bind_ephemeral();
+    let (http_addr, http_listener) = bind_ephemeral();
+    let mut config = base_config("review-post-retry", grpc_addr, http_addr);
+    config.model_discovery_listen_addr = model_addr;
+    config.proxy_transport.quic.request_timeout = Duration::from_millis(300);
+    let listeners = BoundStargateListeners::from_prebound(
+        &config,
+        grpc_listener,
+        model_listener,
+        http_listener,
+        None,
+    )
+    .unwrap();
+    let discovery = SelfDiscovery::new("review-post-retry", grpc_addr, http_addr);
+    let handle = StargateRuntime::new(config, Box::new(discovery), listeners, None)
+        .start()
+        .await
+        .unwrap();
+    let mut fixture = ProxyFixture::new(grpc_addr, http_addr, handle);
+    let tunnel = start_quic_http_tunnel(QuicHttpTunnelConfig::new(
+        "127.0.0.1:0".parse().unwrap(),
+        format!("http://{backend_addr}"),
+    ))
+    .await
+    .unwrap();
+    fixture.register(
+        active_registration_config_with_state(
+            grpc_addr,
+            "review-backend",
+            "",
+            format!("quic://{}", tunnel.listen_addr()),
+            format!("http://{backend_addr}"),
+            active_runtime(model),
+        ),
+        "review registration failed",
+    );
+    fixture.own_tunnel(tunnel);
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while fixture
+            .handle
+            .state()
+            .list_active_models(None, &[model.to_string()])
+            .await
+            .is_empty()
+        {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+
+    let response = fixture
+        .chat_request(model, "one-user-request")
+        .timeout(Duration::from_secs(5))
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.text().await.unwrap();
+    let accepted = hits.load(Ordering::SeqCst);
+    fixture.shutdown().await;
+    backend_task.abort();
+    let _ = backend_task.await;
+    assert_eq!(status, StatusCode::BAD_GATEWAY);
+    assert_eq!(
+        accepted, 1,
+        "one POST was accepted {accepted} times; final status={status}, body={body}"
+    );
+}

src/libraries/rust/stargate/docs/diagrams/chat-completions-e2e.puml

@@ -65,6 +65,14 @@ QuicProxy -> TunnelSvr: send_header(\n  x-method: POST,\n  x-path: /v1/chat/comp
 QuicProxy -> TunnelSvr: send_body(request body chunks)
 QuicProxy -> TunnelSvr: finish (half-close)
 
+note over Proxy,QuicProxy
+  Transport failures may be retried before body submission.
+  After submission, a transport failure returns an error:
+  the backend may already have applied the POST.
+  Explicit retryable pylon rejections can still be replayed
+  within the configured retry limits.
+end note
+
 == Upstream ==
 TunnelSvr -> Upstream: POST /v1/chat/completions\n(reconstructed HTTP request)
 Upstream --> TunnelSvr: HTTP 200\nContent-Type: text/event-stream\n\ndata: {"choices":[...]}\ndata: {"choices":[...]}\ndata: [DONE]