Signals: calls to TypeSafe paced under its rate limits, a 429 honoured, and the wait traced, counted and drawn (#277)

TypeSafe allows 80 requests and 100,000 input tokens a second. Six calls
in flight was a cap on concurrency, not rate: at 74 ms a call it is 81 a
second. Every call to Jev now takes a turn, a 60th of a second or its
guessed tokens' share of 80,000, so up to ten can be in flight. A 429
holds every call back for Retry-After and ends the pass. Each jev span
records paced_ms, /metrics ipx_jev_paced_seconds_total and a rate_limited
result, and the dashboard draws calls and tokens a second against the
limit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -23,10 +23,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A TypeSafe key is checked with TypeSafe before it is saved, and one it refuses is not saved (#271).
 - A new Signal reads the last week instead of the last 30 days, unless an admin sets otherwise (#276).
 - Editing a Signal's name, words or description asks again only about what it had picked out, and about new items as they come, instead of the whole window again (#276).
-- Signals fill faster and cost less: six calls to TypeSafe at once instead of one, everyone's Signals on an item in one call on the server's key, and an item's text cut at 1,000 characters instead of 2,000 (#275).
+- Signals fill faster and cost less: up to ten calls to TypeSafe at once instead of one, paced under TypeSafe's limits of 80 calls and 100,000 input tokens a second (#277), everyone's Signals on an item in one call on the server's key, and an item's text cut at 1,000 characters instead of 2,000 (#275).
 
 ### Fixed
 
+- A 429 from TypeSafe holds every call back as it asks, instead of the pass going on asking (#277).
 - Looking for items to read for Signals read every item on the server on each scan; it now reads only recent ones (#274).
 
 ## [1.3.0] - 2026-10-16

crates/ipx/src/metrics.rs

@@ -46,8 +46,10 @@ pub struct Stats {
     /// When the Signals judge last finished a pass, Unix seconds; 0 for not since starting.
     pub signals_last_pass: AtomicI64,
     /// Calls to Jev by how they ended, in `JEV_RESULTS`' order, and their microseconds all told.
-    pub jev_calls: [AtomicU64; 4],
+    pub jev_calls: [AtomicU64; 5],
     pub jev_micros: AtomicU64,
+    /// Microseconds calls waited for their turn under TypeSafe's limits (#277), part of `jev_micros`.
+    pub jev_paced_micros: AtomicU64,
     /// Tokens TypeSafe counted, input and output, as its answers say.
     pub jev_tokens: [AtomicU64; 2],
     /// The worker's job now, and since when.
@@ -77,8 +79,9 @@ impl Stats {
             signals_judged: AtomicU64::new(0),
             signals_hits: AtomicU64::new(0),
             signals_last_pass: AtomicI64::new(0),
-            jev_calls: [const { AtomicU64::new(0) }; 4],
+            jev_calls: [const { AtomicU64::new(0) }; 5],
             jev_micros: AtomicU64::new(0),
+            jev_paced_micros: AtomicU64::new(0),
             jev_tokens: [const { AtomicU64::new(0) }; 2],
             job: std::sync::Mutex::new(None),
             jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()),
@@ -107,8 +110,9 @@ impl Drop for Job {
 }
 
 /// How a call to Jev ended: answered, its key refused (401-403), the request refused as
-/// malformed (400, 422), or anything else, TypeSafe's 5xx and timeouts.
-pub const JEV_RESULTS: [&str; 4] = ["ok", "key_refused", "malformed", "error"];
+/// malformed (400, 422), over TypeSafe's rate limits (429), or anything else, TypeSafe's 5xx and
+/// timeouts.
+pub const JEV_RESULTS: [&str; 5] = ["ok", "key_refused", "malformed", "rate_limited", "error"];
 
 /// Counts one more of something in progress until dropped, however it ends.
 pub struct Busy(&'static AtomicI64);
@@ -252,7 +256,9 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li
     let labels: Vec<String> = JEV_RESULTS.iter().map(|r| format!("result=\"{r}\"")).collect();
     let calls: Vec<(&str, f64)> = labels.iter().zip(&st.jev_calls).map(|(l, c)| (l.as_str(), c.load(Relaxed) as f64)).collect();
     counter(out, "ipx_jev_requests_total", "Calls to TypeSafe's Jev, Signals' and their icons', by how they ended.", &calls);
-    counter(out, "ipx_jev_seconds_total", "Seconds spent waiting on Jev.", &[("", st.jev_micros.load(Relaxed) as f64 / 1e6)]);
+    counter(out, "ipx_jev_seconds_total", "Seconds spent on calls to Jev, their wait for a turn included.", &[("", st.jev_micros.load(Relaxed) as f64 / 1e6)]);
+    counter(out, "ipx_jev_paced_seconds_total", "Seconds calls to Jev waited for their turn under TypeSafe's rate limits.",
+        &[("", st.jev_paced_micros.load(Relaxed) as f64 / 1e6)]);
     counter(out, "ipx_jev_tokens_total", "Tokens TypeSafe counted for calls to Jev, what they cost.",
         &[("kind=\"input\"", st.jev_tokens[0].load(Relaxed) as f64), ("kind=\"output\"", st.jev_tokens[1].load(Relaxed) as f64)]);
     for (name, help, v) in [

crates/ipx/src/signals.rs

@@ -25,8 +25,14 @@ const BATCH: i64 = 50;
 /// Between passes when nothing wakes it: a scan or a new Signal does, so this is a backstop.
 const IDLE: std::time::Duration = std::time::Duration::from_secs(300);
 /// Calls to Jev in flight at once (#275). One at a time, about 150 ms each, a new Signal's month
-/// of items took twenty minutes; few enough at once to stay a polite client of TypeSafe.
-const IN_FLIGHT: usize = 6;
+/// of items took twenty minutes. Enough for `MAX_RATE` at that pace; the pace is `pace`'s to keep.
+const IN_FLIGHT: usize = 10;
+/// TypeSafe's default limits are 80 requests and 100,000 input tokens a second (#277). Calls are
+/// paced under both, with room left for the icon picks and key checks that share them.
+const MAX_RATE: f64 = 60.0;
+const MAX_TOKENS: f64 = 80_000.0;
+/// When the next call may start, for every call ipx makes to Jev.
+static NEXT_CALL: tokio::sync::Mutex<Option<tokio::time::Instant>> = tokio::sync::Mutex::const_new(None);
 /// How sure Jev has to be that an item is about the keywords.
 const THRESHOLD: f64 = 0.5;
 /// Of an item's text, which is a podcast's whole show notes at times. 2000 until #275: on 55 real
@@ -141,6 +147,13 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
                         tracing::warn!(user = first.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signals request as malformed");
                     }
                 }
+                // Over TypeSafe's limits: what is in flight finishes, nothing more starts, and
+                // the next pass waits for `back_off`.
+                Err(e) if rate_limited(&e) => {
+                    if !stopped.swap(true, Relaxed) {
+                        tracing::warn!(error = %format!("{e:#}"), "TypeSafe asked Signals to slow down");
+                    }
+                }
                 // Left unjudged, so it is asked again next pass.
                 Err(e) => tracing::warn!(user = first.user_id, feed = %first.feed_id, error = %format!("{e:#}"), "Jev could not judge an item"),
             }
@@ -209,6 +222,42 @@ pub fn refuses_key(e: &anyhow::Error) -> bool {
     matches!(status(e), Some(401 | 402 | 403))
 }
 
+fn rate_limited(e: &anyhow::Error) -> bool {
+    status(e) == Some(429)
+}
+
+/// A call's turn: it starts at `next` or now, whichever is later, and holds the line for its
+/// share of a second under both limits. Gives when it starts and when the one after may.
+fn slot(next: tokio::time::Instant, now: tokio::time::Instant, tokens: f64) -> (tokio::time::Instant, tokio::time::Instant) {
+    let start = next.max(now);
+    (start, start + std::time::Duration::from_secs_f64((1.0 / MAX_RATE).max(tokens / MAX_TOKENS)))
+}
+
+/// Waits for this call's turn. Its tokens are guessed from the request, a token to three bytes
+/// of JSON, on the high side: TypeSafe says the true count only in the answer.
+async fn pace(body: &serde_json::Value) {
+    let tokens = body.to_string().len() as f64 / 3.0;
+    let start = {
+        let mut next = NEXT_CALL.lock().await;
+        let now = tokio::time::Instant::now();
+        let (start, after) = slot(next.unwrap_or(now), now, tokens);
+        *next = Some(after);
+        start
+    };
+    let waited = start.saturating_duration_since(tokio::time::Instant::now());
+    tokio::time::sleep_until(start).await;
+    // How long the limits held this call back: near nothing unless ipx is near them (#277).
+    STATS.jev_paced_micros.fetch_add(waited.as_micros() as u64, Relaxed);
+    tracing::Span::current().record("paced_ms", waited.as_millis() as u64);
+}
+
+/// TypeSafe said too many: nobody calls again until it says, or a second.
+async fn back_off(retry_after: Option<u64>) {
+    let until = tokio::time::Instant::now() + std::time::Duration::from_secs(retry_after.unwrap_or(1).min(60));
+    let mut next = NEXT_CALL.lock().await;
+    *next = Some(next.map_or(until, |n| n.max(until)));
+}
+
 fn malformed(e: &anyhow::Error) -> bool {
     matches!(status(e), Some(400 | 422))
 }
@@ -226,6 +275,7 @@ async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result<Vec<Vec<(i64
         "otel.status_code" = tracing::field::Empty,
         input_tokens = tracing::field::Empty,
         output_tokens = tracing::field::Empty,
+        paced_ms = tracing::field::Empty,
     );
     let answer = called(span, ask(ctx, key, &request(group[0], &asks))).await?;
     // A question Jev left unanswered stays unjudged and is asked again.
@@ -288,6 +338,7 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr
         "otel.status_code" = tracing::field::Empty,
         input_tokens = tracing::field::Empty,
         output_tokens = tracing::field::Empty,
+        paced_ms = tracing::field::Empty,
     );
     called(span, async {
         let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?;
@@ -306,6 +357,7 @@ pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> {
         "otel.status_code" = tracing::field::Empty,
         input_tokens = tracing::field::Empty,
         output_tokens = tracing::field::Empty,
+        paced_ms = tracing::field::Empty,
     );
     let body = serde_json::json!({
         "model": "jev-latest",
@@ -324,9 +376,13 @@ async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result<serde_jso
     if let Some(tp) = ipx_directory::otel::traceparent() {
         req = req.header("traceparent", tp);
     }
+    pace(body).await;
     let r = req.send().await?;
     let status = r.status().as_u16();
     tracing::Span::current().record("http.response.status_code", status);
+    if status == 429 {
+        back_off(r.headers().get("retry-after").and_then(|v| v.to_str().ok()?.parse().ok())).await;
+    }
     if !r.status().is_success() {
         let body: String = r.text().await.unwrap_or_default().chars().take(500).collect();
         return Err(Refused { status, body }.into());
@@ -362,7 +418,8 @@ fn outcome<T>(r: &Result<T>) -> usize {
         Ok(_) => 0,
         Err(e) if refuses_key(e) => 1,
         Err(e) if malformed(e) => 2,
-        Err(_) => 3,
+        Err(e) if rate_limited(e) => 3,
+        Err(_) => 4,
     }
 }
 
@@ -477,6 +534,7 @@ mod tests {
         assert_eq!(named(refused(402)), "key_refused");
         assert_eq!(named(refused(422)), "malformed");
         assert_eq!(named(refused(503)), "error");
+        assert_eq!(named(refused(429)), "rate_limited");
         assert_eq!(named(Err(anyhow::anyhow!("timed out"))), "error");
     }
 
@@ -495,6 +553,21 @@ mod tests {
         assert!(r["questions"]["signal_10"].is_object() && r["questions"]["signal_20"].is_object());
     }
 
+    #[test]
+    fn calls_are_paced_under_both_of_typesafes_limits() {
+        let t0 = tokio::time::Instant::now();
+        let ms = |a: tokio::time::Instant| (a - t0).as_secs_f64() * 1000.0;
+        // A small call holds the line for a 60th of a second, under 80 requests a second.
+        let (start, next) = slot(t0, t0, 1000.0);
+        assert_eq!((ms(start), ms(next).round()), (0.0, 17.0));
+        // One of 8,000 tokens, a tenth of a second, under 100,000 tokens a second.
+        let (start, next) = slot(next, t0, 8000.0);
+        assert_eq!((ms(start).round(), ms(next).round()), (17.0, 117.0));
+        // After a quiet spell, a call starts at once.
+        let later = t0 + std::time::Duration::from_secs(5);
+        assert_eq!(slot(next, later, 1000.0).0, later);
+    }
+
     #[test]
     fn a_batch_is_full_by_its_rows_not_its_items() {
         let item = |n: usize| Unjudged {