cortex/frontend/
telemetry.rs

1// Copyright 2015-2025 Deyan Ginev. See the LICENSE
2// file at the top-level directory of this distribution.
3//
4// Licensed under the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>.
5// This file may not be copied, modified, or distributed
6// except according to those terms.
7
8//! Telemetry-dashboard capability: a read-only, retroactive rollup of the per-job `telemetry.json`
9//! the latexml-oxide `cortex_worker` writes into every result archive, across a completed
10//! `(corpus, service)` run.
11//!
12//! Follows the symmetry contract — one shared [`crate::telemetry::TelemetrySummary`] is the read
13//! model for both the agent API (`GET /api/telemetry/<corpus>/<service>`) and the server-rendered
14//! human screen ([`telemetry_report_page`], `GET /telemetry/<corpus>/<service>`). Both live at
15//! top-level `/telemetry` / `/api/telemetry` prefixes (like the `/document/...` report screens) so
16//! neither collides with the same-shape `/corpus/<c>/<s>/<severity>` report routes. Because a
17//! summary aggregates thousands of result
18//! archives off disk (a multi-second scan), it is memoized in an in-memory [`TelemetryCache`] with
19//! a short TTL rather than recomputed per request. The heavy read happens **off** both the cache
20//! lock and the pooled DB connection.
21
22use std::collections::HashMap;
23use std::sync::{Arc, Mutex};
24use std::time::{Duration, Instant};
25
26use rocket::http::Status;
27use rocket::serde::json::Json;
28use rocket::{Route, State};
29use rocket_dyn_templates::{Template, context};
30
31use crate::backend::DbPool;
32use crate::models::{Corpus, Service, Task};
33use crate::telemetry::TelemetrySummary;
34
35/// How long a computed telemetry summary stays fresh before the next view recomputes it (5
36/// minutes).
37const TELEMETRY_TTL: Duration = Duration::from_secs(300);
38
39/// The cache's inner map: `(corpus_id, service_id)` → (when it was computed, the shared summary).
40type TelemetryCacheMap = HashMap<(i32, i32), (Instant, Arc<TelemetrySummary>)>;
41
42/// In-memory TTL cache of computed telemetry summaries, keyed by `(corpus_id, service_id)`. A
43/// summary aggregates every result archive of a completed run (a multi-second disk scan over
44/// thousands of ZIPs), so it is memoized for [`TELEMETRY_TTL`] rather than recomputed per request.
45/// A plain `Mutex<HashMap>` suffices — the entry set is tiny (one per viewed scope) and writes are
46/// rare; the heavy aggregation runs outside the lock.
47#[derive(Default)]
48pub struct TelemetryCache(Mutex<TelemetryCacheMap>);
49
50/// Resolves a `(corpus, service)` name pair to its records, mapping each miss to `404`.
51fn resolve(
52  corpus: &str,
53  service: &str,
54  connection: &mut diesel::PgConnection,
55) -> Result<(Corpus, Service), Status> {
56  let corpus = Corpus::find_by_name(corpus, connection).map_err(|_| Status::NotFound)?;
57  let service = Service::find_by_name(service, connection).map_err(|_| Status::NotFound)?;
58  Ok((corpus, service))
59}
60
61/// Returns the cached telemetry summary for a `(corpus, service)`, recomputing it on a cold/stale
62/// miss. On a miss it resolves the pair, enumerates the completed run's task entries, releases the
63/// pooled connection, and aggregates the result-archive telemetry off disk before caching. `404` on
64/// an unknown corpus/service, `503` if the pool is exhausted.
65fn cached_summary(
66  corpus: &str,
67  service: &str,
68  pool: &State<DbPool>,
69  cache: &State<TelemetryCache>,
70) -> Result<Arc<TelemetrySummary>, Status> {
71  let mut connection = pool.get().map_err(|_| Status::ServiceUnavailable)?;
72  let (corpus, service) = resolve(corpus, service, &mut connection)?;
73  let key = (corpus.id, service.id);
74  // Fast path: a fresh cache hit serves without touching disk. A poisoned lock is a `500`, never a
75  // panic on the request path (DESIGN_PRINCIPLES).
76  {
77    let cached = cache.0.lock().map_err(|_| Status::InternalServerError)?;
78    if let Some((computed_at, summary)) = cached.get(&key)
79      && computed_at.elapsed() < TELEMETRY_TTL
80    {
81      return Ok(summary.clone());
82    }
83  }
84  // Cold/stale: enumerate the entries, then release the pooled connection BEFORE the multi-second
85  // ZIP scan so the aggregation never pins a pooled connection (P-2/P-4 discipline).
86  let entries = Task::completed_entries(corpus.id, service.id, &mut connection)
87    .map_err(|_| Status::InternalServerError)?;
88  drop(connection);
89  let summary = Arc::new(crate::telemetry::aggregate(
90    &corpus.name,
91    &service.name,
92    corpus.sandbox_id(),
93    entries,
94  ));
95  // Store under the lock (last writer wins; a concurrent miss recomputing the same scope is
96  // acceptable and rare).
97  cache
98    .0
99    .lock()
100    .map_err(|_| Status::InternalServerError)?
101    .insert(key, (Instant::now(), summary.clone()));
102  Ok(summary)
103}
104
105/// The telemetry rollup as an agent API (the JSON twin of [`telemetry_report_page`]): wall/RSS
106/// percentiles, per-phase P99, outcome mix, and witness papers for a completed `(corpus, service)`
107/// run. Served from the shared [`TelemetryCache`]. `404` on an unknown corpus/service, `503` if the
108/// pool is exhausted.
109#[get("/api/telemetry/<corpus>/<service>")]
110pub fn api_telemetry(
111  corpus: &str,
112  service: &str,
113  pool: &State<DbPool>,
114  cache: &State<TelemetryCache>,
115) -> Result<Json<TelemetrySummary>, Status> {
116  // The cache hands back a shared `Arc`; clone the (small) inner summary to hand serde an owned
117  // value — `serde` isn't built with the `rc` feature, so `Arc<T>` itself isn't `Serialize`.
118  let summary = cached_summary(corpus, service, pool, cache)?;
119  Ok(Json((*summary).clone()))
120}
121
122/// The telemetry dashboard **screen** (HTML twin of [`api_telemetry`]): the same rolled-up run
123/// telemetry, rendered as an outcome mix, wall/RSS percentile tables, a per-phase P99 bar list, and
124/// the slowest / highest-RSS witness papers (each linking into its per-article forensics). `404` on
125/// an unknown corpus/service, `503` if the pool is exhausted.
126#[get("/telemetry/<corpus>/<service>")]
127pub fn telemetry_report_page(
128  corpus: &str,
129  service: &str,
130  pool: &State<DbPool>,
131  cache: &State<TelemetryCache>,
132) -> Result<Template, Status> {
133  let summary = cached_summary(corpus, service, pool, cache)?;
134  let global = serde_json::json!({
135    "title": format!("Telemetry · {}/{}", summary.corpus, summary.service),
136    "description": "Per-run conversion telemetry: latency, memory, and per-phase profile",
137  });
138  // Hand the template an owned clone of the summary (serde has no `rc` feature, so the shared `Arc`
139  // isn't `Serialize`); the summary is small, and telemetry views are cache-served and infrequent.
140  let summary = (*summary).clone();
141  Ok(Template::render(
142    "telemetry-report",
143    context! { global, summary },
144  ))
145}
146
147/// The route set for the telemetry-dashboard capability (the agent API + the human screen).
148pub fn routes() -> Vec<Route> { routes![api_telemetry, telemetry_report_page] }
149
150#[cfg(test)]
151mod tests {
152  use crate::telemetry::{
153    MathStats, OutcomeWall, PHASES, Percentiles, RssBuckets, TailStats, TelemetrySummary,
154  };
155  use rocket_dyn_templates::tera::{Context, Tera, Value};
156  use std::collections::HashMap;
157
158  /// Render the real `layout` + `telemetry-report` templates against a representative summary, so a
159  /// Tera syntax error or a context-shape mismatch in the hand-written template is caught by the
160  /// suite rather than only at first request. Runs from the repo root (CWD-coupled, like the rest
161  /// of the frontend).
162  #[test]
163  fn telemetry_report_template_renders() {
164    let layout = std::fs::read_to_string("templates/layout.html.tera")
165      .expect("layout template present (run tests from the repo root)");
166    let page = std::fs::read_to_string("templates/telemetry-report.html.tera")
167      .expect("telemetry-report template present");
168    let mut tera = Tera::default();
169    tera
170      .add_raw_templates(vec![
171        ("layout", layout.as_str()),
172        ("telemetry-report", page.as_str()),
173      ])
174      .expect("templates parse and their inheritance resolves");
175    // The layout/report use the app's custom `group_thousands` filter; a passthrough stand-in keeps
176    // this a pure template-shape test (the filter itself is covered in `server.rs`).
177    tera.register_filter(
178      "group_thousands",
179      |value: &Value, _: &HashMap<String, Value>| Ok(value.clone()),
180    );
181
182    let summary = TelemetrySummary {
183      corpus: "arxiv".to_string(),
184      service: "tex_to_html".to_string(),
185      sample_count: 3,
186      skipped: 1,
187      outcome_counts: vec![
188        ("no_problem".to_string(), 2),
189        ("warning".to_string(), 1),
190        ("error".to_string(), 0),
191        ("fatal".to_string(), 0),
192      ],
193      wall_ms: Percentiles {
194        p50: 100,
195        p90: 200,
196        p99: 300,
197        max: 400,
198      },
199      rss_mib: Percentiles {
200        p50: 10,
201        p90: 20,
202        p99: 30,
203        max: 40,
204      },
205      phase_p99_ms: PHASES.iter().map(|phase| (phase.to_string(), 5)).collect(),
206      phase_wall_pct: PHASES
207        .iter()
208        .map(|phase| (phase.to_string(), 100.0 / 17.0))
209        .collect(),
210      tail: TailStats {
211        top1pct_wall_share: 10.0,
212        top5pct_wall_share: 27.0,
213        over_30s: 5,
214        over_60s: 1,
215        over_120s: 0,
216        over_180s: 0,
217      },
218      rss_buckets: RssBuckets {
219        over_2gib: 3,
220        over_3gib: 1,
221        over_4gib: 0,
222      },
223      math: MathStats {
224        formulae: 100,
225        parse_invocations: 90,
226        parse_count: 120,
227        parses_per_formula: 1.33,
228      },
229      slow_tail_dominant: vec![("math_parse".to_string(), 22), ("digest".to_string(), 17)],
230      fatal_profile: OutcomeWall {
231        n: 1,
232        median_ms: 3000,
233        mean_ms: 13000,
234        p99_ms: 98000,
235      },
236      no_problem_profile: OutcomeWall {
237        n: 2,
238        median_ms: 100,
239        mean_ms: 150,
240        p99_ms: 300,
241      },
242      slowest: Some(("1234.5678".to_string(), 400)),
243      highest_rss: Some(("2345.6789".to_string(), 40)),
244      total_formulae: 100,
245      total_graphics_assets: 5,
246      total_output_bytes: 999,
247      generated_unix: 0,
248    };
249    let mut context = Context::new();
250    context.insert(
251      "global",
252      &serde_json::json!({ "title": "t", "description": "d" }),
253    );
254    context.insert("summary", &summary);
255    // Optional layout-only vars, inserted so the shared layout never trips on an undefined lookup.
256    context.insert("is_admin", &false);
257    context.insert("message", &"");
258    context.insert("history", &false);
259
260    let html = tera
261      .render("telemetry-report", &context)
262      .expect("the telemetry-report template renders with a representative summary");
263    assert!(html.contains("Outcome mix"), "renders the outcome section");
264    assert!(
265      html.contains("Per-phase profile"),
266      "renders the phase profile"
267    );
268    assert!(
269      html.contains("/document/arxiv/tex_to_html/1234.5678"),
270      "the slowest witness links to its per-article forensics"
271    );
272    assert!(
273      html.contains("2345.6789"),
274      "renders the highest-RSS witness"
275    );
276    assert!(
277      html.contains("Where wall time goes"),
278      "renders the phase budget"
279    );
280    assert!(
281      html.contains("Wall by outcome"),
282      "renders the outcome wall profiles"
283    );
284    assert!(
285      html.contains("candidate parses/formula"),
286      "renders the math over-parse multiplier"
287    );
288    assert!(
289      html.contains("Slowest-50 dominated by"),
290      "renders the slow-tail driver"
291    );
292    assert!(
293      html.contains("alloc wall"),
294      "renders the RSS pressure buckets"
295    );
296  }
297}
cortex/frontend/telemetry.rs

cortex/frontend/
telemetry.rs
