Stop a stream that generates too slowly

CHANGELOG.md

@@ -17,12 +17,16 @@ for the providers are recognised by value, whatever their format. The
 console shows the rules' names and descriptions as thinkwatch-core
 writes them, in both languages. A value recognised by mistake can be made
 an exception without switching its rule off, and long conversations are
-no longer scanned from the start every turn. The thinkwatch-core crates
+no longer scanned from the start every turn. A provider can set the
+slowest generation its streams may keep up, and a stream that falls
+below it is cut so the client can try again. The thinkwatch-core crates
 move from v0.83.1 to v0.86.0.
 
-No new setting or column. The redaction policy gains one optional field
-(`exceptions`), with endpoints of its own under
-`/api/admin/security/redact/exceptions`.
+No new setting. Two new provider columns
+(`providers.min_speed_tokens_per_sec`, `providers.min_speed_window_secs`,
+added on startup) and one new metric (`gateway_provider_too_slow_total`).
+The redaction policy gains one optional field (`exceptions`), with
+endpoints of its own under `/api/admin/security/redact/exceptions`.
 
 ### Read before upgrading
 
@@ -89,6 +93,24 @@ No new setting or column. The redaction policy gains one optional field
   it and scans only what is new. What is found is the same as without it.
   The cache is rebuilt whenever the rules, the providers' keys or the
   exceptions change.
+- **Stop a stream that generates too slowly** (the provider's
+  "Stop when generation is slower than … tokens/s for … s" setting,
+  `min_speed_tokens_per_sec` 1–1000 and `min_speed_window_secs` 5–120,
+  default 20, on the provider endpoints; blank or `null` is off, the
+  default). From the first token on, a stream is judged once a second:
+  when the text it streamed over the last window (answer, reasoning and
+  tool arguments, four ASCII characters or one other character to a
+  token; Gemini's running count when higher) comes to fewer tokens per
+  second than the minimum, the stream ends with a 503 error in the
+  client's format (`overloaded_error` for Anthropic, `response.failed`
+  for Responses, `UNAVAILABLE` for Gemini), the upstream connection is
+  closed, and the route is charged a failure. Pauses of 5 s or more are
+  left out of the window, and a window that is more than half pauses is
+  not judged. The session's affinity to that route is dropped, the
+  answer so far is billed as for any stream cut short, and the log row
+  carries the error `UpstreamTooSlow` and the speed at the cut
+  (`tokens_per_sec`). Whole answers and Responses WebSocket turns are
+  not measured.
 
 ### Changed
 

crates/common/src/dto/mod.rs

@@ -257,6 +257,15 @@ pub struct CreateProviderRequest {
     #[serde(default)]
     pub headers: Vec<ProviderHeader>,
     pub config: Option<serde_json::Value>,
+    /// Cut a stream that generates fewer tokens per second than this
+    /// (1–1000), averaged over `min_speed_window_secs`. Absent or `null`
+    /// ⇒ off.
+    #[serde(default)]
+    pub min_speed_tokens_per_sec: Option<i64>,
+    /// The window the speed is averaged over, in seconds (5–120).
+    /// Absent ⇒ 20.
+    #[serde(default)]
+    pub min_speed_window_secs: Option<i64>,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone, utoipa::ToSchema)]

crates/common/src/models/provider.rs

@@ -15,6 +15,12 @@ pub struct Provider {
     pub config_json: serde_json::Value,
     pub created_at: DateTime<Utc>,
     pub deleted_at: Option<DateTime<Utc>>,
+    /// The slowest generation a stream from this upstream may keep up,
+    /// in tokens per second (1–1000), averaged over the last
+    /// `min_speed_window_secs` seconds (5–120). `None` ⇒ off. See
+    /// `think_watch_gateway::proxy::min_speed`.
+    pub min_speed_tokens_per_sec: Option<i32>,
+    pub min_speed_window_secs: i32,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]

crates/gateway/src/error.rs

@@ -103,6 +103,21 @@ pub enum GatewayError {
          conversation."
     )]
     SealedHistoryUnavailable { provider: String, model: String },
+    /// A stream generated slower than its provider's minimum (see
+    /// `proxy::min_speed`): `rate` tokens per second over the last `secs`
+    /// seconds, below `min`. The stream was cut; maps to 503, the
+    /// upstream failing, so the client can try again and the route's
+    /// breaker counts it.
+    #[error(
+        "Upstream `{provider}` generated {rate} tokens per second over the last {secs} seconds, \
+         below the minimum of {min}."
+    )]
+    UpstreamTooSlow {
+        provider: String,
+        rate: u32,
+        min: u32,
+        secs: u32,
+    },
 }
 
 impl GatewayError {
@@ -126,6 +141,7 @@ impl GatewayError {
             GatewayError::PolicyBlocked(_) => 403,
             GatewayError::SealedHistoryRefused { status, .. } => i64::from(*status),
             GatewayError::SealedHistoryUnavailable { .. } => 409,
+            GatewayError::UpstreamTooSlow { .. } => 503,
         }
     }
 
@@ -146,6 +162,7 @@ impl GatewayError {
             GatewayError::PolicyBlocked(_) => "PolicyBlocked",
             GatewayError::SealedHistoryRefused { .. } => "SealedHistoryRefused",
             GatewayError::SealedHistoryUnavailable { .. } => "SealedHistoryUnavailable",
+            GatewayError::UpstreamTooSlow { .. } => "UpstreamTooSlow",
         }
     }
 
@@ -256,4 +273,25 @@ mod tests {
         assert_eq!(down.retry_after_secs(), Some(30));
         assert_eq!(down.should_retry(), None);
     }
+
+    #[test]
+    fn a_stream_too_slow_is_the_upstream_failing_and_says_how_slow() {
+        let e = GatewayError::UpstreamTooSlow {
+            provider: "local-vllm".into(),
+            rate: 3,
+            min: 15,
+            secs: 20,
+        };
+        assert_eq!(e.status_code(), 503);
+        assert_eq!(e.error_tag(), "UpstreamTooSlow");
+        assert!(crate::proxy::upstream_failed(
+            e.error_tag(),
+            e.status_code()
+        ));
+        assert_eq!(
+            e.to_string(),
+            "Upstream `local-vllm` generated 3 tokens per second over the last 20 seconds, \
+             below the minimum of 15."
+        );
+    }
 }

crates/gateway/src/lifecycle/mod.rs

@@ -34,6 +34,7 @@ use tw_dialect::ir::Dialect;
 
 use crate::guards::Guards;
 use crate::proxy::generate::{Wire, priced, tokens};
+use crate::proxy::min_speed::{CHECK_EVERY, MinSpeed, SpeedWatch};
 use crate::proxy::retry::{KEEP_ALIVE, KEEP_ALIVE_FRAME};
 use crate::proxy::shaper::{StreamShaper, rewrite_model};
 use crate::proxy::{
@@ -141,6 +142,9 @@ pub(crate) struct ChatPickedRoute {
     /// The route's capacity caps: its token cap counts the answer's
     /// tokens in `record_usage`.
     pub caps: crate::route_caps::RouteCaps,
+    /// The slowest generation the provider's streams may keep up (see
+    /// `proxy::min_speed`).
+    pub min_speed: Option<MinSpeed>,
 }
 
 /// Everything the post-invoke hooks read. Built once before the upstream
@@ -188,6 +192,9 @@ pub(crate) struct Sent {
     /// The attempt under way sent a different reasoning effort than the
     /// caller asked (see `proxy::effort`).
     pub effort: Option<crate::proxy::effort::EffortChange>,
+    /// The stream was cut for generating too slowly, at this many tokens
+    /// per second (see `proxy::min_speed`).
+    pub too_slow: Option<u32>,
 }
 
 /// Build the streaming pump: forward the upstream's bytes to the caller —
@@ -251,6 +258,7 @@ pub(crate) fn build_chat_pump(
     guards: &Guards,
     request: &ChatRequestSnapshot,
     provider: &str,
+    min_speed: Option<MinSpeed>,
 ) -> (axum::response::Response, StreamTail) {
     let readers = Arc::new(Mutex::new(Readers::default()));
     let readers_for_tail = Arc::clone(&readers);
@@ -288,6 +296,7 @@ pub(crate) fn build_chat_pump(
         last_sent: std::time::Instant::now(),
         waiting: false,
         effort: None,
+        too_slow: None,
     }));
     // Gemini clients get no keep-alive comments: the google-genai SDK reads
     // every SSE line as JSON.
@@ -318,7 +327,7 @@ pub(crate) fn build_chat_pump(
             }
             let sent_at = std::time::Instant::now();
             if let Ok(mut s) = sent.lock() {
-                *s = Sent { retries: resend.budget.retries(), last_sent: sent_at, waiting: false, effort: None };
+                *s = Sent { retries: resend.budget.retries(), last_sent: sent_at, waiting: false, effort: None, too_slow: None };
             }
             // What an attempt that failed had read is not this one's: its
             // usage is not billed, its error not the request's.
@@ -392,16 +401,47 @@ pub(crate) fn build_chat_pump(
             // While a failure could still be sent again, the upstream's
             // opening is held back until its first content (see `Opening`).
             let mut hold = resend.budget.open().then(|| Opening::new(wire.dialect));
+            // The provider's slowest generation, judged once a second from
+            // the first token on (see `proxy::min_speed`).
+            let mut speed = min_speed.map(|m| SpeedWatch::new(m, wire.dialect));
+            let mut judge = tokio::time::interval(CHECK_EVERY);
+            judge.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
             let mut source = upstream.bytes_stream();
             loop {
                 let next = tokio::select! {
-                    item = source.next() => Some(item),
-                    _ = tokio::time::sleep_until(quiet_since + KEEP_ALIVE), if hold.is_some() && keep_alive => None,
+                    item = source.next() => Pulled::Item(item),
+                    _ = tokio::time::sleep_until(quiet_since + KEEP_ALIVE), if hold.is_some() && keep_alive => Pulled::KeepAlive,
+                    _ = judge.tick(), if speed.as_ref().is_some_and(SpeedWatch::started) => Pulled::Judge,
                 };
-                let Some(item) = next else {
-                    quiet_since = tokio::time::Instant::now();
-                    yield Ok(Bytes::from_static(KEEP_ALIVE_FRAME));
-                    continue;
+                let item = match next {
+                    Pulled::Item(item) => item,
+                    Pulled::KeepAlive => {
+                        quiet_since = tokio::time::Instant::now();
+                        yield Ok(Bytes::from_static(KEEP_ALIVE_FRAME));
+                        continue;
+                    }
+                    Pulled::Judge => {
+                        let Some((rate, min)) = speed.as_mut().and_then(|s| {
+                            Some((s.too_slow(std::time::Instant::now())?, s.min()))
+                        }) else {
+                            continue;
+                        };
+                        // Too slow: the client is told in its own format, and
+                        // returning drops the upstream's connection.
+                        let err = too_slow(&provider, rate, min);
+                        let mut out = shaper.process(&error_frame(client, err.status_code(), &err.to_string()));
+                        out.extend(shaper.finish());
+                        if let Ok(mut s) = sent.lock() {
+                            s.too_slow = Some(rate);
+                        }
+                        tell(&mut done_tx, StreamOutcome::UpstreamError {
+                            error_type: err.error_tag().to_string(),
+                            message: err.to_string(),
+                            status_code: err.status_code(),
+                        });
+                        yield Ok(Bytes::from(out));
+                        return;
+                    }
                 };
                 let Some(item) = item else {
                     // The upstream's stream ended. One whose opening is still
@@ -448,6 +488,9 @@ pub(crate) fn build_chat_pump(
                 };
                 match item {
                     Ok(chunk) => {
+                        if let Some(s) = speed.as_mut() {
+                            s.feed(&chunk, std::time::Instant::now());
+                        }
                         let (sealed, said) = match readers.lock() {
                             Ok(mut r) => {
                                 r.feed(&chunk);
@@ -691,6 +734,7 @@ pub(crate) fn build_chat_pump(
             last_sent: started_at,
             waiting: false,
             effort: None,
+            too_slow: None,
         });
         let invoked = Invoked {
             client_ip: identity.ip_address.clone(),
@@ -1108,6 +1152,7 @@ fn resending(
             last_sent: std::time::Instant::now(),
             waiting: true,
             effort: None,
+            too_slow: None,
         };
     }
     tracing::info!(
@@ -1224,6 +1269,37 @@ impl Opening {
     }
 }
 
+/// What the pump waits on next: a piece of the upstream's stream (or its
+/// end), a keep-alive due, or a second to judge the stream's speed.
+enum Pulled<T> {
+    Item(T),
+    KeepAlive,
+    Judge,
+}
+
+/// A stream cut for generating at `rate` tokens per second, below `min`:
+/// the error, and the log line saying so.
+fn too_slow(provider: &str, rate: u32, min: MinSpeed) -> GatewayError {
+    tracing::warn!(
+        provider,
+        tokens_per_sec = rate,
+        min_tokens_per_sec = min.tokens_per_sec,
+        window_secs = min.window_secs,
+        "upstream generated too slowly, cutting the stream"
+    );
+    metrics::counter!(
+        "gateway_provider_too_slow_total",
+        "provider" => crate::metrics_labels::normalize_provider_label(provider),
+    )
+    .increment(1);
+    GatewayError::UpstreamTooSlow {
+        provider: provider.to_string(),
+        rate,
+        min: min.tokens_per_sec,
+        secs: min.window_secs,
+    }
+}
+
 /// Tell the tail how the stream ended. Only the first outcome counts.
 fn tell(done_tx: &mut Option<tokio::sync::oneshot::Sender<StreamOutcome>>, outcome: StreamOutcome) {
     if let Some(tx) = done_tx.take() {
@@ -1533,12 +1609,15 @@ impl Surface for ChatCompletionSurface {
             deps.request.request_started_at.elapsed().as_millis() as i64,
             logged_status,
             with_usage_detail(
-                effort_detail(
-                    retried_detail(
-                        pinned_detail(error_detail, deps.route.pinned),
-                        deps.route.sel_record.retries,
+                speed_detail(
+                    effort_detail(
+                        retried_detail(
+                            pinned_detail(error_detail, deps.route.pinned),
+                            deps.route.sel_record.retries,
+                        ),
+                        deps.route.sel_record.effort,
                     ),
-                    deps.route.sel_record.effort,
+                    deps.route.sel_record.too_slow,
                 ),
                 &usage,
                 usage_estimated(&invoked.view),
@@ -1623,6 +1702,24 @@ fn effort_detail(
     Some(serde_json::Value::Object(d))
 }
 
+/// The audit detail, with `"tokens_per_sec"` when the stream was cut for
+/// generating too slowly (see `proxy::min_speed`): its speed over the
+/// window at the cut. A failed request otherwise records no speed.
+fn speed_detail(
+    detail: Option<serde_json::Value>,
+    too_slow: Option<u32>,
+) -> Option<serde_json::Value> {
+    let Some(rate) = too_slow else {
+        return detail;
+    };
+    let mut d = match detail {
+        Some(serde_json::Value::Object(d)) => d,
+        _ => serde_json::Map::new(),
+    };
+    d.insert("tokens_per_sec".into(), rate.into());
+    Some(serde_json::Value::Object(d))
+}
+
 /// The usage a captured view bills — shared by `record_usage` and
 /// `emit_audit` so the budget and the audit row can never disagree.
 fn extract_usage(view: &CapturedView<ChatCompletionSurface>) -> tw_dialect::usage::Usage {
@@ -1852,6 +1949,7 @@ mod tests {
             last_sent: std::time::Instant::now(),
             waiting: false,
             effort: None,
+            too_slow: None,
         });
         resending(
             "up",

crates/gateway/src/proxy/generate.rs

@@ -1233,6 +1233,7 @@ async fn run(
             upstream_model: route.upstream_model.clone(),
             sel_record,
             caps: crate::route_caps::RouteCaps::of(route),
+            min_speed: route.upstream.min_speed,
         },
         cache_enabled: surface.caches,
     };