Signals' metrics by person and whose key paid (#278)

With the server paying for everyone's Signals, its bill had no names.
ipx_signals_judged_total, ipx_signals_hits_total and ipx_jev_tokens_total
now carry user (the account's name) and key (server or own); a call
shared on the server's key is split by each person's share of its
questions. Icon picks bill the person on the server's key, key checks on
their own. The dashboard has tokens and items by person and a Who used
what table over its range.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-17 15:26:50 +00:00
parent ce78599ab4
commit 4eba746ba4
8 changed files with 107 additions and 32 deletions

View File

@@ -109,9 +109,9 @@ from 2026-10-11 15:15 UTC. All gauges:
| `ipx_feed_retries` | feeds waiting to retry a failed name lookup | | `ipx_feed_retries` | feeds waiting to retry a failed name lookup |
| `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_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_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_signals_judged_total{user,key}`, `ipx_signals_hits_total{user,key}` | counters: items Jev judged for Signals, and Signals they fit, by person and whose key paid (`server` or `own`, #278) (#254) |
| `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_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_jev_tokens_total{kind="input"\|"output",user,key}` | counter: tokens TypeSafe counted, as its answers say, by person and whose key paid: what Signals, icon picks and key checks cost (#275, #278). A call shared on the server's key is split by each person's share of its questions |
| `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 | | `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) | | `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) |
| `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the healthcheck last counted it (every 30 s) | | `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the healthcheck last counted it (every 30 s) |

View File

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended, the time they took and the tokens TypeSafe counted, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270, #275). - `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended, the time they took and the tokens TypeSafe counted, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270, #275).
- An admin setting for how many days back a new Signal reads (#276). - An admin setting for how many days back a new Signal reads (#276).
- `/metrics` breaks Signals' items, fits and TypeSafe tokens out by person and by whose key paid, so the server's bill can be split (#278).
- An admin setting, Signals for everyone on the server's TypeSafe key, so the server pays for Jev instead of each person (#271). - An admin setting, Signals for everyone on the server's TypeSafe key, so the server pays for Jev instead of each person (#271).
### Changed ### Changed

View File

@@ -2596,6 +2596,8 @@ pub struct Ask {
#[derive(Debug)] #[derive(Debug)]
pub struct Unjudged { pub struct Unjudged {
pub user_id: i64, pub user_id: i64,
/// Their account's name, what /metrics bills by (#278).
pub user_name: String,
/// Their own key; None when the server's is used for everyone. /// Their own key; None when the server's is used for everyone.
pub key: Option<String>, pub key: Option<String>,
pub feed_id: String, pub feed_id: String,
@@ -2743,7 +2745,7 @@ impl Db {
// rows of signal_verdicts. The same placeholders serve both. // rows of signal_verdicts. The same placeholders serve both.
let half = |items: &str| { let half = |items: &str| {
format!( format!(
"SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.typesafe_key, "SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.name AS user_name, u.typesafe_key,
e.feed_id, e.guid, e.title, e.description, f.title AS feed_title, coalesce(e.published, e.first_seen) AS at e.feed_id, e.guid, e.title, e.description, f.title AS feed_title, coalesce(e.published, e.first_seen) AS at
FROM signals g FROM signals g
JOIN users u ON u.id = g.user_id JOIN users u ON u.id = g.user_id
@@ -2781,6 +2783,7 @@ impl Db {
Some(u) if (u.user_id, &u.feed_id, &u.guid) == (user_id, &feed_id, &guid) => u.signals.push(signal), Some(u) if (u.user_id, &u.feed_id, &u.guid) == (user_id, &feed_id, &guid) => u.signals.push(signal),
_ => out.push(Unjudged { _ => out.push(Unjudged {
user_id, user_id,
user_name: r.try_get("", "user_name")?,
key: r.try_get("", "typesafe_key")?, key: r.try_get("", "typesafe_key")?,
feed_id, feed_id,
guid, guid,

View File

@@ -40,9 +40,6 @@ pub struct Stats {
pub feeds: AtomicI64, pub feeds: AtomicI64,
pub pending: AtomicI64, pub pending: AtomicI64,
pub downloaded: AtomicI64, pub downloaded: AtomicI64,
/// Items Jev has judged for Signals, and Signals they fit (#254).
pub signals_judged: AtomicU64,
pub signals_hits: AtomicU64,
/// When the Signals judge last finished a pass, Unix seconds; 0 for not since starting. /// When the Signals judge last finished a pass, Unix seconds; 0 for not since starting.
pub signals_last_pass: AtomicI64, pub signals_last_pass: AtomicI64,
/// Calls to Jev by how they ended, in `JEV_RESULTS`' order, and their microseconds all told. /// Calls to Jev by how they ended, in `JEV_RESULTS`' order, and their microseconds all told.
@@ -50,8 +47,10 @@ pub struct Stats {
pub jev_micros: AtomicU64, pub jev_micros: AtomicU64,
/// Microseconds calls waited for their turn under TypeSafe's limits (#277), part of `jev_micros`. /// Microseconds calls waited for their turn under TypeSafe's limits (#277), part of `jev_micros`.
pub jev_paced_micros: AtomicU64, pub jev_paced_micros: AtomicU64,
/// Tokens TypeSafe counted, input and output, as its answers say. /// What each person's Signals, icons and key checks used, and whose key paid (#278): items
pub jev_tokens: [AtomicU64; 2], /// judged, Signals they fit, and TypeSafe's input and output tokens. Floats: a call shared by
/// several people on the server's key is split between them.
people: std::sync::Mutex<std::collections::BTreeMap<(String, &'static str), [f64; 4]>>,
/// The worker's job now, and since when. /// The worker's job now, and since when.
job: std::sync::Mutex<Option<(&'static str, std::time::Instant)>>, job: std::sync::Mutex<Option<(&'static str, std::time::Instant)>>,
/// Jobs finished, by kind: how many, and their seconds all told. /// Jobs finished, by kind: how many, and their seconds all told.
@@ -76,18 +75,26 @@ impl Stats {
feeds: AtomicI64::new(-1), feeds: AtomicI64::new(-1),
pending: AtomicI64::new(-1), pending: AtomicI64::new(-1),
downloaded: AtomicI64::new(-1), downloaded: AtomicI64::new(-1),
signals_judged: AtomicU64::new(0),
signals_hits: AtomicU64::new(0),
signals_last_pass: AtomicI64::new(0), signals_last_pass: AtomicI64::new(0),
jev_calls: [const { AtomicU64::new(0) }; 5], jev_calls: [const { AtomicU64::new(0) }; 5],
jev_micros: AtomicU64::new(0), jev_micros: AtomicU64::new(0),
jev_paced_micros: AtomicU64::new(0), jev_paced_micros: AtomicU64::new(0),
jev_tokens: [const { AtomicU64::new(0) }; 2], people: std::sync::Mutex::new(std::collections::BTreeMap::new()),
job: std::sync::Mutex::new(None), job: std::sync::Mutex::new(None),
jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()), jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()),
} }
} }
/// Adds to what `who` used on `key` ("server" or "own"): items judged, fits, input and output
/// tokens.
pub fn bill(&self, who: &str, key: &'static str, used: [f64; 4]) {
let mut people = self.people.lock().unwrap();
let row = people.entry((who.to_owned(), key)).or_default();
for (r, u) in row.iter_mut().zip(used) {
*r += u;
}
}
/// Marks the worker as on a job of `kind` until what it returns is dropped, however the job /// Marks the worker as on a job of `kind` until what it returns is dropped, however the job
/// ends: finished, failed or stopped. /// ends: finished, failed or stopped.
pub fn job(&'static self, kind: &'static str) -> Job { pub fn job(&'static self, kind: &'static str) -> Job {
@@ -247,8 +254,16 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li
counter(out, "ipx_art_evicted_total", "Images thrown out to keep the artwork under its limit.", &[("", st.art_evicted.load(Relaxed) as f64)]); counter(out, "ipx_art_evicted_total", "Images thrown out to keep the artwork under its limit.", &[("", st.art_evicted.load(Relaxed) as f64)]);
counter(out, "ipx_art_requests_total", "Artwork asked for: served from disk, or fetched from the publisher.", counter(out, "ipx_art_requests_total", "Artwork asked for: served from disk, or fetched from the publisher.",
&[("result=\"hit\"", st.art_hits.load(Relaxed) as f64), ("result=\"miss\"", st.art_misses.load(Relaxed) as f64)]); &[("result=\"hit\"", st.art_hits.load(Relaxed) as f64), ("result=\"miss\"", st.art_misses.load(Relaxed) as f64)]);
counter(out, "ipx_signals_judged_total", "Items Jev has judged for someone's Signals.", &[("", st.signals_judged.load(Relaxed) as f64)]); // By person and whose key paid (#278). An account's name is often an email address: /metrics
counter(out, "ipx_signals_hits_total", "Signals an item was found to fit.", &[("", st.signals_hits.load(Relaxed) as f64)]); // is on the LAN only, as Prometheus is.
let people = st.people.lock().unwrap().clone();
let labels: Vec<String> = people.keys().map(|(who, key)| format!("user=\"{}\",key=\"{key}\"", label(who))).collect();
let by = |n: usize| labels.iter().zip(people.values()).map(|(l, v)| (l.as_str(), v[n])).collect::<Vec<_>>();
counter(out, "ipx_signals_judged_total", "Items Jev has judged for someone's Signals, by person and whose key paid.", &by(0));
counter(out, "ipx_signals_hits_total", "Signals an item was found to fit, by person and whose key paid.", &by(1));
let tokens: Vec<String> = labels.iter().flat_map(|l| [format!("kind=\"input\",{l}"), format!("kind=\"output\",{l}")]).collect();
let used: Vec<(&str, f64)> = tokens.iter().zip(people.values().flat_map(|v| [v[2], v[3]])).map(|(l, v)| (l.as_str(), v)).collect();
counter(out, "ipx_jev_tokens_total", "Tokens TypeSafe counted for calls to Jev, what they cost, by person and whose key paid.", &used);
let last = st.signals_last_pass.load(Relaxed); let last = st.signals_last_pass.load(Relaxed);
if last > 0 { if last > 0 {
gauge(out, "ipx_signals_last_pass_age_seconds", "Seconds since the Signals judge last finished a pass.", &[("", (now - last) as f64)]); gauge(out, "ipx_signals_last_pass_age_seconds", "Seconds since the Signals judge last finished a pass.", &[("", (now - last) as f64)]);
@@ -259,8 +274,6 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li
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_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.", 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)]); &[("", 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 [ for (name, help, v) in [
("ipx_feeds", "Feeds in the catalogue, as last counted.", &st.feeds), ("ipx_feeds", "Feeds in the catalogue, as last counted.", &st.feeds),
("ipx_downloads_pending", "Files a scan will download on its own, as last counted.", &st.pending), ("ipx_downloads_pending", "Files a scan will download on its own, as last counted.", &st.pending),
@@ -273,6 +286,11 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li
} }
} }
/// A label's value as Prometheus's text format reads it.
fn label(v: &str) -> String {
v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n")
}
/// A counter, which only goes up while the daemon runs, in Prometheus's text format. /// A counter, which only goes up while the daemon runs, in Prometheus's text format.
pub fn counter(out: &mut String, name: &str, help: &str, samples: &[(&str, f64)]) { pub fn counter(out: &mut String, name: &str, help: &str, samples: &[(&str, f64)]) {
write(out, "counter", name, help, samples); write(out, "counter", name, help, samples);
@@ -328,6 +346,14 @@ mod tests {
assert!(out.contains("ipx_jobs_total{kind=\"scan\"} 1\n") && out.contains("# TYPE ipx_jobs_total counter"), "{out}"); assert!(out.contains("ipx_jobs_total{kind=\"scan\"} 1\n") && out.contains("# TYPE ipx_jobs_total counter"), "{out}");
assert!(out.contains("ipx_event_queue 3\n") && out.contains("ipx_feed_retries 1\n")); assert!(out.contains("ipx_event_queue 3\n") && out.contains("ipx_feed_retries 1\n"));
assert!(out.contains("ipx_jev_requests_total{result=\"key_refused\"} 0\n") && !out.contains("ipx_signals_last_pass_age_seconds"), "{out}"); assert!(out.contains("ipx_jev_requests_total{result=\"key_refused\"} 0\n") && !out.contains("ipx_signals_last_pass_age_seconds"), "{out}");
T.bill("sam@example.net", "server", [1.0, 1.0, 600.0, 50.0]);
T.bill("sam@example.net", "server", [1.0, 0.0, 300.5, 25.0]);
T.bill("a \"b\"", "own", [0.0, 0.0, 10.0, 1.0]);
let mut out = String::new();
work(&mut out, &T, 0, 0, 0);
assert!(out.contains("ipx_signals_judged_total{user=\"sam@example.net\",key=\"server\"} 2\n"), "{out}");
assert!(out.contains("ipx_jev_tokens_total{kind=\"input\",user=\"sam@example.net\",key=\"server\"} 900.5\n"), "{out}");
assert!(out.contains("ipx_jev_tokens_total{kind=\"output\",user=\"a \\\"b\\\"\",key=\"own\"} 1\n"), "{out}");
// The library's counts are left out until the daemon has counted it. // The library's counts are left out until the daemon has counted it.
assert!(!out.contains("ipx_feeds ")); assert!(!out.contains("ipx_feeds "));
} }

View File

@@ -88,6 +88,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
let span = tracing::info_span!("signals", entries = items.len(), calls = tracing::field::Empty, judged = tracing::field::Empty, hits = tracing::field::Empty); let span = tracing::info_span!("signals", entries = items.len(), calls = tracing::field::Empty, judged = tracing::field::Empty, hits = tracing::field::Empty);
async { async {
let groups = calls(&items, server.is_some()); let groups = calls(&items, server.is_some());
let paid = if server.is_some() { "server" } else { "own" };
tracing::Span::current().record("calls", groups.len()); tracing::Span::current().record("calls", groups.len());
let refused = std::sync::Mutex::new(HashSet::new()); let refused = std::sync::Mutex::new(HashSet::new());
let stopped = AtomicBool::new(false); let stopped = AtomicBool::new(false);
@@ -106,7 +107,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
return None; return None;
} }
let key = server.as_deref().or(first.key.as_deref())?; let key = server.as_deref().or(first.key.as_deref())?;
Some((group, judge(ctx, group, key).await)) Some((group, judge(ctx, group, key, paid).await))
} }
}) })
.buffer_unordered(IN_FLIGHT); .buffer_unordered(IN_FLIGHT);
@@ -118,8 +119,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
for (item, verdicts) in group.iter().zip(verdicts) { for (item, verdicts) in group.iter().zip(verdicts) {
judged += 1; judged += 1;
let hits = verdicts.iter().filter(|(_, h)| *h).count(); let hits = verdicts.iter().filter(|(_, h)| *h).count();
STATS.signals_judged.fetch_add(1, Relaxed); STATS.bill(&item.user_name, paid, [1.0, hits as f64, 0.0, 0.0]);
STATS.signals_hits.fetch_add(hits as u64, Relaxed);
if hits > 0 { if hits > 0 {
*hits_for.entry(item.user_id).or_default() += hits; *hits_for.entry(item.user_id).or_default() += hits;
tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal"); tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal");
@@ -263,7 +263,7 @@ fn malformed(e: &anyhow::Error) -> bool {
} }
/// One call about one item, for everyone in `group`: their verdicts, in the group's order. /// One call about one item, for everyone in `group`: their verdicts, in the group's order.
async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result<Vec<Vec<(i64, bool)>>> { async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str, paid: &'static str) -> Result<Vec<Vec<(i64, bool)>>> {
let asks: Vec<&Ask> = group.iter().flat_map(|i| &i.signals).collect(); let asks: Vec<&Ask> = group.iter().flat_map(|i| &i.signals).collect();
let span = tracing::info_span!( let span = tracing::info_span!(
"jev", "jev",
@@ -278,6 +278,12 @@ async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result<Vec<Vec<(i64
paced_ms = tracing::field::Empty, paced_ms = tracing::field::Empty,
); );
let answer = called(span, ask(ctx, key, &request(group[0], &asks))).await?; let answer = called(span, ask(ctx, key, &request(group[0], &asks))).await?;
// A shared call is billed by each person's share of its questions, which are most of it.
let [input, output] = usage(&answer);
for item in group {
let share = item.signals.len() as f64 / asks.len() as f64;
STATS.bill(&item.user_name, paid, [0.0, 0.0, input * share, output * share]);
}
// A question Jev left unanswered stays unjudged and is asked again. // A question Jev left unanswered stays unjudged and is asked again.
Ok(group.iter().map(|i| i.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect()).collect()) Ok(group.iter().map(|i| i.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect()).collect())
} }
@@ -329,7 +335,7 @@ fn hit(answer: &serde_json::Value, id: i64) -> Option<bool> {
/// chooses among options and cannot name one, so the page searches Font Awesome, whose index is /// chooses among options and cannot name one, so the page searches Font Awesome, whose index is
/// ipx-web's, and sends the likeliest. On the server's key, not the Signal owner's: it is the /// ipx-web's, and sends the likeliest. On the server's key, not the Signal owner's: it is the
/// server's page doing it, once a Signal. /// server's page doing it, once a Signal.
pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> Result<String> { pub async fn pick_icon(ctx: &Ctx, who: &str, key: &str, name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> Result<String> {
let span = tracing::info_span!( let span = tracing::info_span!(
"jev_icon", "jev_icon",
candidates = candidates.len(), candidates = candidates.len(),
@@ -342,6 +348,8 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr
); );
called(span, async { called(span, async {
let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?; let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?;
let [input, output] = usage(&answer);
STATS.bill(who, "server", [0.0, 0.0, input, output]);
chosen(&answer, candidates).ok_or_else(|| anyhow::anyhow!("Jev chose no icon offered: {answer}")) chosen(&answer, candidates).ok_or_else(|| anyhow::anyhow!("Jev chose no icon offered: {answer}"))
}) })
.await .await
@@ -349,7 +357,7 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr
/// Whether TypeSafe takes `key`, by one short question, before it is saved (#271): a key it /// Whether TypeSafe takes `key`, by one short question, before it is saved (#271): a key it
/// refuses would sit saved with nothing ever picked out, and nobody told why. /// refuses would sit saved with nothing ever picked out, and nobody told why.
pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> { pub async fn check_key(ctx: &Ctx, who: &str, key: &str) -> Result<()> {
let span = tracing::info_span!( let span = tracing::info_span!(
"jev_key_check", "jev_key_check",
"http.response.status_code" = tracing::field::Empty, "http.response.status_code" = tracing::field::Empty,
@@ -365,7 +373,10 @@ pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> {
"questions": { "ok": { "type": "noul", "instructions": "Is this a key check?", "questions": { "ok": { "type": "noul", "instructions": "Is this a key check?",
"criteria": { "true": "It is.", "false": "It is not." } } }, "criteria": { "true": "It is.", "false": "It is not." } } },
}); });
called(span, ask(ctx, key, &body)).await.map(|_| ()) let answer = called(span, ask(ctx, key, &body)).await?;
let [input, output] = usage(&answer);
STATS.bill(who, "own", [0.0, 0.0, input, output]);
Ok(())
} }
/// Puts `body` to Jev on `key` and gives its answer, or what TypeSafe refused it with. Carries /// Puts `body` to Jev on `key` and gives its answer, or what TypeSafe refused it with. Carries
@@ -388,16 +399,21 @@ async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result<serde_jso
return Err(Refused { status, body }.into()); return Err(Refused { status, body }.into());
} }
let answer: serde_json::Value = r.json().await?; let answer: serde_json::Value = r.json().await?;
// What a call cost, which TypeSafe says and nothing else would (#275). // What a call cost, which TypeSafe says and nothing else would (#275); billed by the caller,
for (n, kind) in ["input_tokens", "output_tokens"].iter().enumerate() { // who knows whose it was (#278).
for kind in ["input_tokens", "output_tokens"] {
if let Some(t) = answer["usage"][kind].as_u64() { if let Some(t) = answer["usage"][kind].as_u64() {
STATS.jev_tokens[n].fetch_add(t, Relaxed); tracing::Span::current().record(kind, t);
tracing::Span::current().record(*kind, t);
} }
} }
Ok(answer) Ok(answer)
} }
/// Input and output tokens, as an answer counts them; 0 when it does not.
fn usage(answer: &serde_json::Value) -> [f64; 2] {
["input_tokens", "output_tokens"].map(|k| answer["usage"][k].as_f64().unwrap_or(0.0))
}
/// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span /// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span
/// failed, which a field named `error` alone does not: Tempo's failed traces left Jev's out. /// failed, which a field named `error` alone does not: Tempo's failed traces left Jev's out.
async fn called<T>(span: tracing::Span, call: impl std::future::Future<Output = Result<T>>) -> Result<T> { async fn called<T>(span: tracing::Span, call: impl std::future::Future<Output = Result<T>>) -> Result<T> {
@@ -541,7 +557,7 @@ mod tests {
#[test] #[test]
fn on_the_servers_key_one_item_is_one_call_for_everyone() { fn on_the_servers_key_one_item_is_one_call_for_everyone() {
let row = |user_id, guid: &str| Unjudged { let row = |user_id, guid: &str| Unjudged {
user_id, key: None, feed_id: "f".into(), guid: guid.into(), feed_title: None, title: None, description: None, user_id, user_name: format!("u{user_id}"), key: None, feed_id: "f".into(), guid: guid.into(), feed_title: None, title: None, description: None,
signals: vec![Ask { id: user_id * 10, name: "s".into(), keywords: vec![], description: None }], signals: vec![Ask { id: user_id * 10, name: "s".into(), keywords: vec![], description: None }],
}; };
let items = [row(1, "x"), row(2, "x"), row(1, "y")]; let items = [row(1, "x"), row(2, "x"), row(1, "y")];
@@ -571,7 +587,7 @@ mod tests {
#[test] #[test]
fn a_batch_is_full_by_its_rows_not_its_items() { fn a_batch_is_full_by_its_rows_not_its_items() {
let item = |n: usize| Unjudged { let item = |n: usize| Unjudged {
user_id: 1, key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None, user_id: 1, user_name: "a".into(), key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None,
signals: (0..n as i64).map(|id| Ask { id, name: "s".into(), keywords: vec![], description: None }).collect(), signals: (0..n as i64).map(|id| Ask { id, name: "s".into(), keywords: vec![], description: None }).collect(),
}; };
assert!(full(&(0..25).map(|_| item(2)).collect::<Vec<_>>()), "25 items of two Signals each is the 50 asked for"); assert!(full(&(0..25).map(|_| item(2)).collect::<Vec<_>>()), "25 items of two Signals each is the 50 asked for");
@@ -582,6 +598,7 @@ mod tests {
fn each_signal_is_its_own_question_on_the_items_words() { fn each_signal_is_its_own_question_on_the_items_words() {
let item = Unjudged { let item = Unjudged {
user_id: 1, user_id: 1,
user_name: "a".into(),
key: Some("k".into()), key: Some("k".into()),
feed_id: "f".into(), feed_id: "f".into(),
guid: "g".into(), guid: "g".into(),

View File

@@ -418,7 +418,7 @@ async fn patch_me(
return Err(ApiError::bad_request("that does not look like a TypeSafe key")); return Err(ApiError::bad_request("that does not look like a TypeSafe key"));
} }
if let Some(k) = &key { if let Some(k) = &key {
crate::signals::check_key(&state.ctx, k).await.map_err(|e| { crate::signals::check_key(&state.ctx, &user.name, k).await.map_err(|e| {
if crate::signals::refuses_key(&e) { if crate::signals::refuses_key(&e) {
ApiError::bad_request("TypeSafe did not accept that key") ApiError::bad_request("TypeSafe did not accept that key")
} else { } else {
@@ -527,7 +527,7 @@ struct IconAsk {
/// or 404 when the server has no key. /// or 404 when the server has no key.
async fn pick_signal_icon( async fn pick_signal_icon(
State(state): State<WebState>, State(state): State<WebState>,
_user: crate::db::User, user: crate::db::User,
Json(body): Json<IconAsk>, Json(body): Json<IconAsk>,
) -> Result<Json<serde_json::Value>, ApiError> { ) -> Result<Json<serde_json::Value>, ApiError> {
let Some(key) = crate::signals::typesafe_key() else { let Some(key) = crate::signals::typesafe_key() else {
@@ -547,7 +547,7 @@ async fn pick_signal_icon(
if candidates.is_empty() || candidates.len() > 60 { if candidates.is_empty() || candidates.len() > 60 {
return Err(ApiError::bad_request("between 1 and 60 icons to choose from")); return Err(ApiError::bad_request("between 1 and 60 icons to choose from"));
} }
let icon = crate::signals::pick_icon(&state.ctx, &key, &name, &words, description.as_deref(), &candidates) let icon = crate::signals::pick_icon(&state.ctx, &user.name, &key, &name, &words, description.as_deref(), &candidates)
.await .await
.map_err(|e| { .map_err(|e| {
tracing::warn!(error = %format!("{e:#}"), "Jev could not pick an icon"); tracing::warn!(error = %format!("{e:#}"), "Jev could not pick an icon");

View File

@@ -392,7 +392,12 @@ erDiagram
- **Keywords changed while an item is being judged** can leave that one item's verdict from the old - **Keywords changed while an item is being judged** can leave that one item's verdict from the old
words. words.
- **The meter is the server's**, `ipx_jev_tokens_total` and `ipx_jev_requests_total` on - **The meter is the server's**, `ipx_jev_tokens_total` and `ipx_jev_requests_total` on
`/metrics`, and the Signals row of the dashboard: tokens as TypeSafe's answers count them. It is `/metrics`, and the Signals row of the dashboard: tokens as TypeSafe's answers count them, by
person (`user`, the account's name) and whose key paid (`key`, `server` or `own`; #278), with
`ipx_signals_judged_total` and `ipx_signals_hits_total` the same way. A call shared by several
people on the server's key is split by each one's share of its questions. Icon picks bill the
person on the server's key, key checks the person on their own. The dashboard's **Who used what**
sums them over its time range. It is
one call per item in the person's feeds as they come, and per item in the window for a new Signal. one call per item in the person's feeds as they come, and per item in the window for a new Signal.
- **Only the window** is judged by publish date, apart from what an edit asks again. - **Only the window** is judged by publish date, apart from what an edit asks again.

View File

@@ -291,6 +291,29 @@ p = ts("Against TypeSafe's limits", [prom('sum(rate(ipx_jev_requests_total{job="
p["fieldConfig"]["defaults"]["custom"]["thresholdsStyle"] = {"mode": "line"} p["fieldConfig"]["defaults"]["custom"]["thresholdsStyle"] = {"mode": "line"}
p["fieldConfig"]["defaults"]["thresholds"] = {"mode": "absolute", "steps": LIMIT(80)} p["fieldConfig"]["defaults"]["thresholds"] = {"mode": "absolute", "steps": LIMIT(80)}
y += 8 y += 8
# By person and whose key paid (#278): who the server's TypeSafe bill is for. An item shared on the
# server's key is split by each person's share of its questions.
ts("Input tokens by person", [prom('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="input"}[$__rate_interval]))',
legend="{{user}} ({{key}} key)")],
0, w=8, bars=True, stack=True, desc="server is the server's TypeSafe key, what the server pays for; own is the person's own key.")
ts("Items judged by person", [prom('sum by (user, key) (increase(ipx_signals_judged_total{job="ipx"}[$__rate_interval]))', legend="{{user}} ({{key}} key)"),
prom('sum by (user) (increase(ipx_signals_hits_total{job="ipx"}[$__rate_interval]))', ref="B", legend="{{user}}: fit a Signal")],
8, w=8, bars=True)
def ptab(expr, ref):
return {"refId": ref, "datasource": PROM, "expr": expr, "instant": True, "format": "table"}
panel("table", "Who used what", 8, 8, 16,
[ptab('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="input"}[$__range]))', "A"),
ptab('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="output"}[$__range]))', "B"),
ptab('sum by (user, key) (increase(ipx_signals_judged_total{job="ipx"}[$__range]))', "C")],
description="Over the dashboard's time range: each person's input and output tokens and items judged, and whose key paid. "
"Icon picks are on the server's key, key checks on the person's own.",
transformations=[{"id": "merge", "options": {}},
{"id": "organize", "options": {"excludeByName": {"Time": True},
"renameByName": {"user": "Person", "key": "Key", "Value #A": "Input tokens",
"Value #B": "Output tokens", "Value #C": "Items judged"}}}],
options={"showHeader": True, "sortBy": [{"displayName": "Input tokens", "desc": True}]},
fieldConfig={"defaults": {"decimals": 0}, "overrides": []})
y += 8
# ---- ipx: the process, what it holds between requests and jobs (#194) # ---- ipx: the process, what it holds between requests and jobs (#194)
row("ipx: process") row("ipx: process")