From ce78599ab4812f70abd93cb21669ca3b95439248 Mon Sep 17 00:00:00 2001 From: rays Date: Sat, 17 Oct 2026 14:38:20 +0000 Subject: [PATCH] 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 --- .claude/skills/ipx-prod-check/SKILL.md | 5 +- crates/ipx/CHANGELOG.md | 3 +- crates/ipx/src/metrics.rs | 16 ++++-- crates/ipx/src/signals.rs | 79 +++++++++++++++++++++++++- docs/signals.md | 11 +++- grafana/dashboard.py | 18 ++++-- 6 files changed, 115 insertions(+), 17 deletions(-) diff --git a/.claude/skills/ipx-prod-check/SKILL.md b/.claude/skills/ipx-prod-check/SKILL.md index 0fe8acc..5395811 100644 --- a/.claude/skills/ipx-prod-check/SKILL.md +++ b/.claude/skills/ipx-prod-check/SKILL.md @@ -110,7 +110,7 @@ from 2026-10-11 15:15 UTC. All gauges: | `ipx_art_cache_bytes`, `ipx_art_cache_limit_bytes` | artwork kept, as of the last sweep, and the most kept: kept sits at the limit by design | | `ipx_art_evicted_total`, `ipx_art_requests_total{result="hit"\|"miss"}` | counters: images the sweep threw out to stay under the limit, and artwork served from disk or fetched from the publisher; many evicted at every sweep with a falling hit rate means the limit is too small | | `ipx_signals_judged_total`, `ipx_signals_hits_total` | counters: items Jev judged for Signals, and Signals they fit (#254) | -| `ipx_jev_requests_total{result="ok"\|"key_refused"\|"malformed"\|"error"}`, `ipx_jev_seconds_total` | counters: calls to TypeSafe's Jev, Signals' and icons', by how they ended, and the seconds waited; `malformed` is ipx's request at fault (#257) | +| `ipx_jev_requests_total{result="ok"\|"key_refused"\|"malformed"\|"rate_limited"\|"error"}`, `ipx_jev_seconds_total`, `ipx_jev_paced_seconds_total` | counters: calls to TypeSafe's Jev, Signals' and icons', by how they ended; their seconds, the wait for a turn included; and that wait alone. `malformed` is ipx's request at fault (#257); `rate_limited` is a 429 from TypeSafe, whose limits are 80 calls and 100,000 input tokens a second, which ipx paces itself under at 60 and 80,000 (#277) | | `ipx_jev_tokens_total{kind="input"\|"output"}` | counter: tokens TypeSafe counted, as its answers say: what Signals cost (#275) | | `ipx_signals_last_pass_age_seconds`, `ipx_signals_keys_refused` | since the Signals judge last finished a pass (at most 5 minutes when well; absent until one has); people whose key TypeSafe refused | | `directory_process_resident_bytes`, `directory_process_open_fds`, `directory_feed_fetches_total`, `directory_feed_bytes_total`, `directory_uptime_seconds`, `directory_build_info` | the directory's, `job="ipx-directory"`, from its `metrics_bind` (production `192.168.1.130:9188`, LAN only) | @@ -216,7 +216,8 @@ a count says something happened, the lines and traces say why. where the worker's time went (reap was most of it until #193). Web requests in flight near the pool's max mean the pool is what they wait on. Live updates missed mean a page showed stale counts until reloaded. DeadAir over a minute on one file is a file ffmpeg chokes on. - Any `malformed` Jev call is ipx asking wrongly (#257); `key_refused` is someone's key; the Signals + Any `rate_limited` Jev call means the pacing is not holding ipx under TypeSafe's limits (#277); each + `jev` span has `paced_ms`, its wait for a turn. Any `malformed` Jev call is ipx asking wrongly (#257); `key_refused` is someone's key; the Signals judge over 10 minutes since a pass is stuck on Jev: its `signals` trace shows which `jev` call. 7. **What the page saw.** The browser's own errors and timings, which no server line shows. diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 62a2086..20cf009 100644 --- a/crates/ipx/CHANGELOG.md +++ b/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 diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index 4b42466..c4567bb 100644 --- a/crates/ipx/src/metrics.rs +++ b/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 = 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 [ diff --git a/crates/ipx/src/signals.rs b/crates/ipx/src/signals.rs index fe3e829..49aeb63 100644 --- a/crates/ipx/src/signals.rs +++ b/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> = 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 { 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) { + 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 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(r: &Result) -> 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 { diff --git a/docs/signals.md b/docs/signals.md index f03a015..eb0be1d 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -93,8 +93,14 @@ is asked by; without it each wake read all of `entries`, 605 MB, 566 ms to find Rows for one item and one person are grouped, so an item is one call whatever the number of Signals. On the server's key, everyone's Signals on the same item go in one call too, its text -sent once (#275). Six calls are in flight at once (`IN_FLIGHT`): one at a time, about 150 ms each, -a month of items took twenty minutes. A full batch means there may be more, and the next pass +sent once (#275). Up to ten calls are in flight at once (`IN_FLIGHT`): one at a time, about 150 ms +each, a month of items took twenty minutes. Every call to Jev, Signals', icons' and key checks', +takes a turn (`pace`, #277): TypeSafe's default limits are 80 requests and 100,000 input tokens a +second, and ipx keeps to 60 and 80,000, each call holding the line for a 60th of a second or its +guessed tokens' share, whichever is longer. A 429 holds every call back for its `Retry-After`, or +a second, and ends the pass. Each `jev` span says how long it waited (`paced_ms`), and +`ipx_jev_paced_seconds_total` adds them up; the dashboard draws calls and tokens a second against +the limits. A full batch means there may be more, and the next pass starts at once. Otherwise the judge sleeps until woken or for 5 minutes. ### Every answer is kept @@ -181,6 +187,7 @@ missing from the answer is left unjudged and asked again next pass. | 401, 402, 403, on the server's key | Everyone's key is refused. The pass ends and the queue waits for the next one, five minutes on. Logged: "TypeSafe refused the server's key for Signals". | | 401, 402, 403, on a person's key | The key is refused. That person is passed over until they save a key or the daemon restarts, so a bad key is not sent the whole queue. Logged once: "TypeSafe refused a Signal key". | | 400, 422 | The request itself is wrong, so every item's would be. The pass ends; the items wait for the next one. Logged: "TypeSafe refused a Signals request as malformed", with TypeSafe's reason. | +| 429 | Over TypeSafe's limits. Every call waits for `Retry-After` or a second; the pass ends, what was in flight kept, and the rest wait for the next. Logged once: "TypeSafe asked Signals to slow down". | | anything else, or no answer | That item is left unjudged and asked again next pass. Logged: "Jev could not judge an item". | The error keeps TypeSafe's reply, cut to 500 characters, for example: diff --git a/grafana/dashboard.py b/grafana/dashboard.py index 4f4d577..05d324b 100644 --- a/grafana/dashboard.py +++ b/grafana/dashboard.py @@ -260,11 +260,12 @@ ts("Items judged", [prom('sum(increase(ipx_signals_judged_total{job="ipx"}[$__ra prom('sum(increase(ipx_signals_hits_total{job="ipx"}[$__rate_interval]))', ref="B", legend="fit a Signal")], 0, w=8, bars=True, desc="Each item is judged once against every Signal of its subscribers that has not had it; a burst follows a scan or a new Signal.") ts("Calls to Jev", [prom('sum by (result) (increase(ipx_jev_requests_total{job="ipx"}[$__rate_interval]))', legend="{{result}}"), - prom('sum(increase(ipx_jev_seconds_total{job="ipx"}[$__rate_interval])) / ' + prom('(sum(increase(ipx_jev_seconds_total{job="ipx"}[$__rate_interval])) - sum(increase(ipx_jev_paced_seconds_total{job="ipx"}[$__rate_interval]))) / ' 'sum(increase(ipx_jev_requests_total{job="ipx"}[$__rate_interval]))', ref="B", legend="seconds a call")], 8, w=8, bars=True, stack=True, desc="key_refused stops that person's Signals until they save another key; malformed stops the pass and is ipx's to fix; " - "error is TypeSafe failing or slow, and the item is asked again next pass.")["fieldConfig"]["overrides"] = [ + "rate_limited is TypeSafe saying slow down (#277), and every call waits; " + "error is TypeSafe failing or slow, and the item is asked again next pass. Seconds a call is TypeSafe's, without the wait for a turn.")["fieldConfig"]["overrides"] = [ {"matcher": {"id": "byName", "options": "seconds a call"}, "properties": [{"id": "unit", "value": "s"}, {"id": "custom.axisPlacement", "value": "right"}, {"id": "custom.drawStyle", "value": "line"}, {"id": "custom.stacking", "value": {"mode": "none"}}]}] @@ -276,10 +277,19 @@ ts("Signals judge", [prom('max(ipx_signals_last_pass_age_seconds{job="ipx"})', l "properties": [{"id": "unit", "value": "short"}, {"id": "custom.axisPlacement", "value": "right"}]}] y += 8 ts("Tokens", [prom('sum by (kind) (increase(ipx_jev_tokens_total{job="ipx"}[$__rate_interval]))', legend="{{kind}}")], - 0, w=12, bars=True, stack=True, desc="What TypeSafe counted for each call, as its answers say: what Signals, icons and key checks cost.") + 0, w=8, bars=True, stack=True, desc="What TypeSafe counted for each call, as its answers say: what Signals, icons and key checks cost.") ts("Tokens a call", [prom('sum by (kind) (increase(ipx_jev_tokens_total{job="ipx"}[$__rate_interval])) / ignoring(kind) group_left ' 'sum(increase(ipx_jev_requests_total{job="ipx", result="ok"}[$__rate_interval]))', legend="{{kind}}")], - 12, w=12, desc="An item's text is most of the input; on the server's key, one call carries everyone's Signals on that item.") + 8, w=8, desc="Most of the input is the questions, one for each Signal, then the item's text, cut at 1,000 characters.") +# TypeSafe's default limits (#277): ipx paces itself to 60 calls and 80,000 input tokens a second. +LIMIT = lambda n: [{"color": "transparent", "value": None}, {"color": "red", "value": n}] +p = ts("Against TypeSafe's limits", [prom('sum(rate(ipx_jev_requests_total{job="ipx"}[$__rate_interval]))', legend="calls a second"), + prom('sum(rate(ipx_jev_tokens_total{job="ipx", kind="input"}[$__rate_interval])) / 1000', ref="B", legend="thousand input tokens a second"), + prom('sum(rate(ipx_jev_paced_seconds_total{job="ipx"}[$__rate_interval]))', ref="C", legend="seconds waited for a turn, a second")], + 16, w=8, desc="TypeSafe allows 80 calls and 100,000 input tokens a second; the red line is 80 on both scales (calls, and thousands of tokens). " + "Waiting for a turn means ipx is holding itself under them, and is the time a new Signal's backfill takes longer for it.") +p["fieldConfig"]["defaults"]["custom"]["thresholdsStyle"] = {"mode": "line"} +p["fieldConfig"]["defaults"]["thresholds"] = {"mode": "absolute", "steps": LIMIT(80)} y += 8 # ---- ipx: the process, what it holds between requests and jobs (#194)