Metrics for the worker's job, the queues and the work in flight (#195)

The healthcheck answers while the worker is stuck on one job, and nothing
showed which job or for how long. STATS, one static of atomics, is kept as
the work happens: the worker's job and since when (a guard, so a failed or
stopped job is counted too), jobs and their seconds by kind, when a scan
last finished, web requests, feed fetches and downloads in progress, bytes
downloaded, live updates dropped for a listener that fell behind, DeadAir's
backlog and how long it has been on its file, and the artwork kept. With
the event queue, the DNS retries and the healthcheck's last counts, /metrics
still reads memory only. The dashboard's Daemon row and the prod-check
skill take them up.

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

.claude/skills/ipx-prod-check/SKILL.md

@@ -68,6 +68,16 @@ from 2026-10-11 15:15 UTC. All gauges:
 | `ipx_runtime_workers`, `ipx_runtime_tasks`, `ipx_runtime_queue` | tokio: threads, live tasks, tasks waiting for a thread |
 | `ipx_process_resident_bytes`, `ipx_process_open_fds` | memory and open files and sockets |
 | `ipx_uptime_seconds`, `ipx_build_info{version}` | since the daemon started; what runs |
+| `ipx_worker_busy`, `ipx_worker_job_seconds{kind}` | the worker's job now (`scan`, `reap`, `download`, `status`, `reload`) and for how long; the healthcheck passes while it is stuck |
+| `ipx_jobs_total{kind}`, `ipx_job_seconds_total{kind}` | counters: jobs finished, and their seconds all told |
+| `ipx_last_scan_age_seconds` | since a scan last finished; absent until one has since starting |
+| `ipx_http_requests_in_flight`, `ipx_scan_feeds_fetching`, `ipx_downloads_active` | work in progress |
+| `ipx_download_bytes_total` | counter; `rate()` is the download speed |
+| `ipx_event_queue`, `ipx_events_missed_total` | live updates unread by some listener; counter of those dropped for one that fell behind |
+| `ipx_deadair_backlog`, `ipx_deadair_busy_seconds` | files of shows with DeadAir on not yet looked at (episodes everyone heard through stay counted); how long it has been on its file |
+| `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 |
+| `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the healthcheck last counted it (every 30 s) |
 
 ## The checks
 
@@ -135,6 +145,12 @@ a count says something happened, the lines and traces say why.
    $Q prom 'ipx_process_resident_bytes{job="ipx"} - ipx_process_resident_bytes{job="ipx"} offset $range' 24h
    $Q prom 'ipx_process_open_fds{job="ipx"} - ipx_process_open_fds{job="ipx"} offset $range' 24h
    $Q prom 'ipx_runtime_tasks{job="ipx"} - ipx_runtime_tasks{job="ipx"} offset $range' 24h
+   $Q prom 'max(max_over_time(ipx_worker_job_seconds{job="ipx"}[$range])) by (kind)' 24h
+   $Q prom 'max(max_over_time(ipx_last_scan_age_seconds{job="ipx"}[$range]))' 24h
+   $Q prom 'sum by (kind) (increase(ipx_job_seconds_total{job="ipx"}[$range]))' 24h
+   $Q prom 'max(max_over_time(ipx_http_requests_in_flight{job="ipx"}[$range]))' 24h
+   $Q prom 'sum(increase(ipx_events_missed_total{job="ipx"}[$range]))' 24h
+   $Q prom 'max(max_over_time(ipx_deadair_busy_seconds{job="ipx"}[$range]))' 24h
    ```
    `up` 0 means Prometheus cannot reach the port: the container is down or `metrics_bind` is
    unset. Connections in use at the pool's max (10 on Postgres) means something is waiting for
@@ -145,6 +161,13 @@ a count says something happened, the lines and traces say why.
    is a leak (a file or socket not closed, a task never ending); a step after a deploy is not. A
    reset of the uptime is a restart: match it with a deploy, or it is a crash (check 4).
 
+   The worker: one job over 30 minutes (`ipx_worker_job_seconds`), or a scan last finished over
+   15 minutes ago, is the worker stuck, which the healthcheck does not notice; its trace and the
+   log lines of that time say on what. A kind whose share of `ipx_job_seconds_total` jumps is
+   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.
+
 Also check the container itself, since Loki cannot see a daemon that is not running:
 
 ```sh

CHANGELOG.md

@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- With `[web] metrics_bind` set, `/metrics` gives Prometheus, on a thread and port of its own, what the daemon holds between requests: database connections in use against the pool's size, commands waiting for the worker, live-update listeners, the runtime's tasks, memory and open files. The iPX dashboard graphs them in a Daemon row (#194).
+- With `[web] metrics_bind` set, `/metrics` gives Prometheus, on a thread and port of its own, what the daemon holds between requests: database connections in use against the pool's size, commands waiting for the worker, live-update listeners, the runtime's tasks, memory and open files. So are the worker's job and how long it has run, jobs done and their time by kind, when a scan last finished, web requests, feed fetches and downloads in progress, download speed, live updates queued and missed, DeadAir's backlog and the file it is on, feeds waiting to retry, the artwork kept and the library's counts. The iPX dashboard graphs them in a Daemon row (#194, #195).
 - The page reports its errors, the player's, how long play takes to start and how long it stalls to the server's log, and the requests one tap sets off share a trace, the playlist, audio and saved place of what it played included (#190, #192).
 - Saving an episode from a show with DeadAir on saves it without its dead air, at your breathing room (#186).
 - Choose the pause DeadAir leaves where it cuts a silence, from 0.2 to 0.5 seconds, for each show in its settings, as its breathing room, on a slider from brisk to relaxed. It is 0.3 seconds unless you change it, up from 0.2, which made speech sound rushed (#182, #183).

grafana/dashboard.py

@@ -163,6 +163,33 @@ y += 8
 ts("Memory", [prom('ipx_process_resident_bytes{job="ipx"}', legend="resident")], 0, w=12, unit="bytes")
 ts("Open files and sockets", [prom('ipx_process_open_fds{job="ipx"}', legend="open")], 12, w=12)
 y += 8
+# Jobs and queues (#195)
+ts("Worker", [prom('max by (kind) (ipx_worker_job_seconds{job="ipx"})', legend="on a {{kind}} for"),
+              prom('max(ipx_last_scan_age_seconds{job="ipx"})', ref="B", legend="since the last scan finished")],
+   0, w=8, unit="s", desc="A job's time that only climbs, or no scan for over 10 minutes, means the worker is stuck; the healthcheck still passes.")
+ts("Jobs done", [prom('sum by (kind) (increase(ipx_jobs_total{job="ipx"}[$__rate_interval]))', legend="{{kind}}")],
+   8, w=8, bars=True, stack=True)
+ts("Time on jobs", [prom('sum by (kind) (increase(ipx_job_seconds_total{job="ipx"}[$__rate_interval]))', legend="{{kind}}")],
+   16, w=8, unit="s", bars=True, stack=True, desc="Seconds the worker spent on each kind of job in each interval.")
+y += 8
+ts("In flight", [prom('max(ipx_http_requests_in_flight{job="ipx"})', legend="web requests"),
+                 prom('max(ipx_scan_feeds_fetching{job="ipx"})', ref="B", legend="feeds fetching"),
+                 prom('max(ipx_downloads_active{job="ipx"})', ref="C", legend="downloads"),
+                 prom('2 - max(ipx_torrent_slots_free{job="ipx"})', ref="D", legend="torrents")],
+   0, w=8)
+ts("Download speed", [prom('sum(rate(ipx_download_bytes_total{job="ipx"}[$__rate_interval]))', legend="downloading")],
+   8, w=8, unit="Bps")
+ts("Backlogs", [prom('max(ipx_deadair_backlog{job="ipx"})', legend="DeadAir to look at"),
+                prom('max(ipx_feed_retries{job="ipx"})', ref="B", legend="feeds to retry a lookup"),
+                prom('max(ipx_event_queue{job="ipx"})', ref="C", legend="live updates unread"),
+                prom('sum(increase(ipx_events_missed_total{job="ipx"}[$__rate_interval]))', ref="D", legend="live updates missed")],
+   16, w=8)
+y += 8
+ts("DeadAir on a file", [prom('max(ipx_deadair_busy_seconds{job="ipx"})', legend="for")], 0, w=12, unit="s",
+   desc="How long DeadAir has been on the file it is analysing; a few seconds each.")
+ts("Artwork kept", [prom('max(ipx_art_cache_bytes{job="ipx"})', legend="kept"),
+                    prom('max(ipx_art_cache_limit_bytes{job="ipx"})', ref="B", legend="the most kept")], 12, w=12, unit="bytes")
+y += 8
 
 # ---- Traces
 row("Traces")

src/art.rs

@@ -97,6 +97,7 @@ pub fn trim(dir: &Path, limit: u64) -> usize {
             gone += 1;
         }
     }
+    crate::metrics::STATS.art_cache_bytes.store(total, std::sync::atomic::Ordering::Relaxed);
     gone
 }
 

src/deadair.rs

@@ -405,11 +405,19 @@ async fn have_ffmpeg() -> bool {
 // them as passed over if a large library makes the pass slow.
 async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result<bool> {
     // Nobody with DeadAir on for anything, which is most passes: nothing more to ask the database.
+    use std::sync::atomic::Ordering::Relaxed;
+    let st = &crate::metrics::STATS;
     let wanting = Wanting::load(ctx, None).await?;
     if wanting.people.is_empty() {
+        st.deadair_backlog.store(0, Relaxed);
         return Ok(false);
     }
-    for t in ctx.db.deadair_todo().await? {
+    let todo = ctx.db.deadair_todo().await?;
+    // For /metrics (#195): files of feeds someone has DeadAir on, from memory, without the
+    // per-file subscription and finished checks below.
+    let wanted = |t: &crate::db::DeadAirTodo| !failed.contains(&t.id) && wanting.people.keys().any(|&u| wanting.wants(u, &t.feed_id));
+    st.deadair_backlog.store(todo.iter().filter(|t| wanted(t)).count() as i64, Relaxed);
+    for t in todo {
         if failed.contains(&t.id) {
             continue;
         }
@@ -427,7 +435,10 @@ async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result<
         // A trace of its own for each file looked at, not for the passes that find nothing to do,
         // which are most of them.
         let span = tracing::info_span!("deadair", feed = t.feed_id, enclosure = t.id, cuts = tracing::field::Empty);
-        let plans = match analyse(&src).instrument(span.clone()).await {
+        st.deadair_busy_since.store(crate::db::now(), Relaxed);
+        let analysed = analyse(&src).instrument(span.clone()).await;
+        st.deadair_busy_since.store(0, Relaxed);
+        let plans = match analysed {
             Ok(p) => p.unwrap_or_default(),
             Err(e) => {
                 tracing::warn!(path = %src.display(), error = %format!("{e:#}"), "DeadAir could not read a file");

src/download.rs

@@ -172,10 +172,12 @@ pub async fn download(
         .with_context(|| format!("creating {}", tmp.display()))?;
     let mut done: u64 = 0;
     let mut stream = resp.bytes_stream();
+    let _active = crate::metrics::busy(&crate::metrics::STATS.downloads_active);
     while let Some(chunk) = stream.next().await {
         let chunk = chunk.context("reading body")?;
         file.write_all(&chunk).await?;
         done += chunk.len() as u64;
+        crate::metrics::STATS.download_bytes.fetch_add(chunk.len() as u64, std::sync::atomic::Ordering::Relaxed);
         on_progress(done, total);
     }
     file.flush().await?;

src/main.rs

@@ -528,11 +528,21 @@ async fn user_cmd(ctx: &Arc<Ctx>, cmd: UserCmd) -> Result<()> {
 }
 
 async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
+    // What the worker is on, for /metrics: the healthcheck answers while it is stuck (#195).
+    let _job = metrics::STATS.job(match &cmd {
+        Cmd::Fetch { .. } => "scan",
+        Cmd::Reap { .. } => "reap",
+        Cmd::Download { .. } => "download",
+        Cmd::Status => "status",
+        Cmd::Reload => "reload",
+    });
     match cmd {
         Cmd::Fetch { feed, force, feeds } => {
             // Make room before pulling more down, as the original did per download.
             reap(ctx, false, false).await?;
-            fetch(ctx, feed.as_deref(), force, &feeds).await
+            fetch(ctx, feed.as_deref(), force, &feeds).await?;
+            metrics::STATS.last_scan_done.store(db::now(), std::sync::atomic::Ordering::Relaxed);
+            Ok(())
         }
         Cmd::Reap { dry_run } => reap(ctx, dry_run, true).await,
         Cmd::Download { enclosure } => download_one(ctx, enclosure).await,
@@ -560,6 +570,12 @@ pub(crate) async fn status(ctx: &Ctx) -> Event {
     match ctx.db.counts().await {
         Ok((pending, downloaded)) => {
             let feeds = subscriptions(ctx).await.map(|s| s.len()).unwrap_or(0);
+            // Kept for /metrics, which asks the database nothing (#195): the healthcheck asks this every 30 s.
+            use std::sync::atomic::Ordering::Relaxed;
+            let st = &metrics::STATS;
+            st.feeds.store(feeds as i64, Relaxed);
+            st.pending.store(pending as i64, Relaxed);
+            st.downloaded.store(downloaded as i64, Relaxed);
             Event::Status { feeds, pending, downloaded }
         }
         Err(e) => Event::Error { msg: format!("{e:#}") },
@@ -1247,6 +1263,7 @@ async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool, scope: &[String]
                     if !early {
                         return None;
                     }
+                    let _fetching = metrics::busy(&metrics::STATS.feeds_fetching);
                     Some(feed::fetch(&client, &feed_cfg, etag.as_deref(), modified.as_deref()).await)
                 },
                 tracing::Span::current(),

src/metrics.rs

@@ -8,6 +8,92 @@
 
 use std::fmt::Write as _;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicI64, AtomicU64, Ordering::Relaxed};
+
+/// What the daemon counts as it works, for /metrics to read without asking anything (#195).
+/// Global: the web layer, the worker, downloads, artwork and DeadAir each add to it, and a handle
+/// threaded through all of them was more code than the counts.
+pub static STATS: Stats = Stats::new();
+
+pub struct Stats {
+    pub http_in_flight: AtomicI64,
+    /// Live updates dropped for a page or socket client that fell behind the broadcast.
+    pub events_missed: AtomicU64,
+    pub downloads_active: AtomicI64,
+    pub download_bytes: AtomicU64,
+    /// Feeds a scan is fetching at this moment, up to its AHEAD.
+    pub feeds_fetching: AtomicI64,
+    /// When a scan last finished, Unix seconds; 0 for not since starting.
+    pub last_scan_done: AtomicI64,
+    /// Files someone wants shortened that DeadAir has not looked at, as of its last pass.
+    pub deadair_backlog: AtomicI64,
+    /// When DeadAir started on the file it is on, Unix seconds; 0 when it is on none.
+    pub deadair_busy_since: AtomicI64,
+    pub art_cache_bytes: AtomicU64,
+    /// The library as the healthcheck last counted it, every 30 s; -1 until it has.
+    pub feeds: AtomicI64,
+    pub pending: AtomicI64,
+    pub downloaded: AtomicI64,
+    /// The worker's job now, and since when.
+    job: std::sync::Mutex<Option<(&'static str, std::time::Instant)>>,
+    /// Jobs finished, by kind: how many, and their seconds all told.
+    jobs: std::sync::Mutex<std::collections::BTreeMap<&'static str, (u64, f64)>>,
+}
+
+impl Stats {
+    pub const fn new() -> Self {
+        Self {
+            http_in_flight: AtomicI64::new(0),
+            events_missed: AtomicU64::new(0),
+            downloads_active: AtomicI64::new(0),
+            download_bytes: AtomicU64::new(0),
+            feeds_fetching: AtomicI64::new(0),
+            last_scan_done: AtomicI64::new(0),
+            deadair_backlog: AtomicI64::new(0),
+            deadair_busy_since: AtomicI64::new(0),
+            art_cache_bytes: AtomicU64::new(0),
+            feeds: AtomicI64::new(-1),
+            pending: AtomicI64::new(-1),
+            downloaded: AtomicI64::new(-1),
+            job: std::sync::Mutex::new(None),
+            jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()),
+        }
+    }
+
+    /// Marks the worker as on a job of `kind` until what it returns is dropped, however the job
+    /// ends: finished, failed or stopped.
+    pub fn job(&'static self, kind: &'static str) -> Job {
+        *self.job.lock().unwrap() = Some((kind, std::time::Instant::now()));
+        Job(self)
+    }
+}
+
+pub struct Job(&'static Stats);
+
+impl Drop for Job {
+    fn drop(&mut self) {
+        if let Some((kind, since)) = self.0.job.lock().unwrap().take() {
+            let mut jobs = self.0.jobs.lock().unwrap();
+            let e = jobs.entry(kind).or_default();
+            e.0 += 1;
+            e.1 += since.elapsed().as_secs_f64();
+        }
+    }
+}
+
+/// Counts one more of something in progress until dropped, however it ends.
+pub struct Busy(&'static AtomicI64);
+
+pub fn busy(gauge: &'static AtomicI64) -> Busy {
+    gauge.fetch_add(1, Relaxed);
+    Busy(gauge)
+}
+
+impl Drop for Busy {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Relaxed);
+    }
+}
 
 /// What the metrics are read from.
 pub struct Sources {
@@ -69,6 +155,7 @@ pub fn render(s: &Sources) -> String {
         &[("", s.ctx.torrent_slots.available_permits() as f64)]);
     gauge(&mut out, "ipx_deadair_playing", "Episodes someone is playing through DeadAir's playlist.",
         &[("", s.playing.lock().map_or(0, |p| p.len()) as f64)]);
+    work(&mut out, &STATS, s.events.len(), s.ctx.retry.lock().map_or(0, |r| r.len()), s.ctx.cfg().general.art_cache_mb * 1_048_576);
     let rt = s.runtime.metrics();
     gauge(&mut out, "ipx_runtime_workers", "The async runtime's worker threads.", &[("", rt.num_workers() as f64)]);
     gauge(&mut out, "ipx_runtime_tasks", "Tasks alive in the async runtime.", &[("", rt.num_alive_tasks() as f64)]);
@@ -86,10 +173,63 @@ pub fn render(s: &Sources) -> String {
     out
 }
 
+/// The work in progress and done, from `STATS`: jobs, queues and what is in flight (#195).
+fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_limit: u64) {
+    let now = crate::db::now();
+    let job = *st.job.lock().unwrap();
+    gauge(out, "ipx_worker_busy", "1 while the worker is on a job.", &[("", job.is_some() as u8 as f64)]);
+    if let Some((kind, since)) = job {
+        let label = format!("kind=\"{kind}\"");
+        gauge(out, "ipx_worker_job_seconds", "How long the worker has been on its job, by the job's kind.",
+            &[(&label, since.elapsed().as_secs_f64())]);
+    }
+    let jobs = st.jobs.lock().unwrap().clone();
+    let labels: Vec<String> = jobs.keys().map(|k| format!("kind=\"{k}\"")).collect();
+    let pairs = |f: fn(&(u64, f64)) -> f64| labels.iter().zip(jobs.values()).map(|(l, v)| (l.as_str(), f(v))).collect::<Vec<_>>();
+    counter(out, "ipx_jobs_total", "Jobs the worker has finished, by kind.", &pairs(|v| v.0 as f64));
+    counter(out, "ipx_job_seconds_total", "Seconds the worker has spent on jobs, by kind.", &pairs(|v| v.1));
+    let last = st.last_scan_done.load(Relaxed);
+    if last > 0 {
+        gauge(out, "ipx_last_scan_age_seconds", "Seconds since a scan last finished.", &[("", (now - last) as f64)]);
+    }
+    gauge(out, "ipx_http_requests_in_flight", "Web requests being answered.", &[("", st.http_in_flight.load(Relaxed) as f64)]);
+    gauge(out, "ipx_scan_feeds_fetching", "Feeds a scan is fetching at once.", &[("", st.feeds_fetching.load(Relaxed) as f64)]);
+    gauge(out, "ipx_downloads_active", "Files downloading now.", &[("", st.downloads_active.load(Relaxed) as f64)]);
+    counter(out, "ipx_download_bytes_total", "Bytes downloaded since the daemon started.", &[("", st.download_bytes.load(Relaxed) as f64)]);
+    gauge(out, "ipx_event_queue", "Live updates sent and not yet read by every listener.", &[("", event_queue as f64)]);
+    counter(out, "ipx_events_missed_total", "Live updates dropped for a listener that fell behind.", &[("", st.events_missed.load(Relaxed) as f64)]);
+    gauge(out, "ipx_deadair_backlog", "Files of shows someone has DeadAir on that it has not looked at, those already heard through included.", &[("", st.deadair_backlog.load(Relaxed) as f64)]);
+    let since = st.deadair_busy_since.load(Relaxed);
+    gauge(out, "ipx_deadair_busy_seconds", "How long DeadAir has been on its file; 0 when on none.",
+        &[("", if since > 0 { (now - since) as f64 } else { 0.0 })]);
+    gauge(out, "ipx_feed_retries", "Feeds waiting to try a failed name lookup again.", &[("", retries as f64)]);
+    gauge(out, "ipx_art_cache_bytes", "Artwork kept on disk, as of the last sweep.", &[("", st.art_cache_bytes.load(Relaxed) as f64)]);
+    gauge(out, "ipx_art_cache_limit_bytes", "The most artwork kept.", &[("", art_limit as f64)]);
+    for (name, help, v) in [
+        ("ipx_feeds", "Feeds in the catalogue, as the healthcheck last counted.", &st.feeds),
+        ("ipx_downloads_pending", "Files a scan will download on its own, as the healthcheck last counted.", &st.pending),
+        ("ipx_files_downloaded", "Files on disk, as the healthcheck last counted.", &st.downloaded),
+    ] {
+        let v = v.load(Relaxed);
+        if v >= 0 {
+            gauge(out, name, help, &[("", v as f64)]);
+        }
+    }
+}
+
+/// 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)]) {
+    write(out, "counter", name, help, samples);
+}
+
 /// One gauge in Prometheus's text format: its help, its type, and a line for each set of labels
 /// (`state="idle"`, or "" for none).
 pub fn gauge(out: &mut String, name: &str, help: &str, samples: &[(&str, f64)]) {
-    let _ = writeln!(out, "# HELP {name} {help}\n# TYPE {name} gauge");
+    write(out, "gauge", name, help, samples);
+}
+
+fn write(out: &mut String, kind: &str, name: &str, help: &str, samples: &[(&str, f64)]) {
+    let _ = writeln!(out, "# HELP {name} {help}\n# TYPE {name} {kind}");
     for (labels, value) in samples {
         let labels = if labels.is_empty() { String::new() } else { format!("{{{labels}}}") };
         let _ = writeln!(out, "{name}{labels} {value}");
@@ -123,6 +263,28 @@ mod tests {
         );
     }
 
+    #[test]
+    fn jobs_and_work_in_progress_are_counted_however_they_end() {
+        static T: Stats = Stats::new();
+        {
+            let _j = T.job("scan");
+            let _b = busy(&T.downloads_active);
+            assert_eq!(T.downloads_active.load(Relaxed), 1);
+            let mut out = String::new();
+            work(&mut out, &T, 0, 0, 0);
+            assert!(out.contains("ipx_worker_busy 1\n") && out.contains("ipx_worker_job_seconds{kind=\"scan\"}"), "{out}");
+        }
+        // Dropped, as an error or a stop drops them: nothing left in progress, one scan done.
+        assert_eq!(T.downloads_active.load(Relaxed), 0);
+        let mut out = String::new();
+        work(&mut out, &T, 3, 1, 10);
+        assert!(out.contains("ipx_worker_busy 0\n") && !out.contains("ipx_worker_job_seconds"), "{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"));
+        // The library's counts are left out until the healthcheck has counted it.
+        assert!(!out.contains("ipx_feeds "));
+    }
+
     #[test]
     fn the_process_reads_its_memory_and_files_on_linux() {
         let (rss, fds) = process();

src/web.rs

@@ -1959,7 +1959,10 @@ async fn events(
                     }
                     return Some((futures_util::stream::iter(out), (rx, state, user)));
                 }
-                Err(broadcast::error::RecvError::Lagged(_)) => {}
+                // Counted for /metrics: a page that misses updates shows stale counts (#195).
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    crate::metrics::STATS.events_missed.fetch_add(n, std::sync::atomic::Ordering::Relaxed);
+                }
                 Err(broadcast::error::RecvError::Closed) => return None,
             }
         }
@@ -2528,7 +2531,9 @@ async fn access_log(req: Request, next: Next) -> Response {
         let _ = span.set_parent(parent);
     }
     let started = std::time::Instant::now();
+    let in_flight = crate::metrics::busy(&crate::metrics::STATS.http_in_flight);
     let resp = tracing::Instrument::instrument(next.run(req), span.clone()).await;
+    drop(in_flight);
     span.record("http.response.status_code", resp.status().as_u16());
     if !quiet {
         let duration_ms = started.elapsed().as_millis() as u64;