mimalloc for ipx and the directory, and metrics and traces for memory (#224)

ipx's resident memory climbed from about 40 MB to 500-850 MB each night while scans ran five
times as often, and stayed there until a restart; the directory sat at 1.25 GiB idle. glibc's
allocator keeps what each thread's arena freed. mimalloc gives freed pages back, so resident
memory now follows what is in use. jemalloc was the first choice for its in-use figures, but its
upstream was archived in 2025.

Declared once in the directory crate, which ipx builds in, so both binaries get it.

What was missing to see it: the directory served no /metrics (now on metrics_bind, a port apart
from the tunnel's), nothing counted the feed bytes fetched, and the fetch span had no body size.
The feed crate's spans never reached Tempo either: both trace exporters filtered to their own
crate, so ipx_feed's fetch and parse spans were dropped.

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

crates/directory/CHANGELOG.md

@@ -10,8 +10,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- With `metrics_bind` set, the directory serves Prometheus metrics on a port of its own: memory, open files, and the feeds it has fetched and their bytes. A feed check's trace says how large the feed was (#224).
+
 ### Changed
 
+- The directory uses mimalloc instead of glibc's allocator, so it gives back memory it has freed instead of holding over a gigabyte while idle (#224).
 - The example in docs/directory.md has no healthcheck; `ipx-directory health` is still there to run by hand (#218).
 
 ## [1.0.0] - 2026-10-12

crates/directory/Cargo.toml

@@ -16,6 +16,7 @@ ipx-feed.workspace = true
 opentelemetry.workspace = true
 opentelemetry-otlp.workspace = true
 opentelemetry_sdk.workspace = true
+mimalloc.workspace = true
 percent-encoding.workspace = true
 reqwest.workspace = true
 sea-orm.workspace = true

crates/directory/src/config.rs

@@ -14,6 +14,10 @@ pub struct Config {
     /// The environment variable holding the database URL, so the file holds no secret.
     #[serde(default = "default_database_url_env")]
     pub database_url_env: String,
+    /// Where to serve /metrics for Prometheus (#224), apart from `bind`, which the tunnel
+    /// publishes: `0.0.0.0:9188`. Empty serves none.
+    #[serde(default)]
+    pub metrics_bind: String,
     #[serde(default)]
     pub tokens: Vec<TokenCfg>,
 }

crates/directory/src/lib.rs

@@ -8,6 +8,7 @@
 //! people's behalf.
 
 pub mod config;
+pub mod metrics;
 pub mod otel;
 mod site;
 pub mod store;

crates/directory/src/main.rs

@@ -82,7 +82,7 @@ async fn serve(path: &std::path::Path) -> Result<()> {
         .with(json.then(|| tracing_subscriber::fmt::layer().json().flatten_event(true).with_writer(std::io::stderr).with_filter(filter())))
         .with(otel.as_ref().map(|p| {
             use opentelemetry::trace::TracerProvider;
-            tracing_opentelemetry::layer().with_tracer(p.tracer("ipx-directory")).with_filter(tracing_subscriber::EnvFilter::new("ipx_directory=info"))
+            tracing_opentelemetry::layer().with_tracer(p.tracer("ipx-directory")).with_filter(tracing_subscriber::EnvFilter::new("ipx_directory=info,ipx_feed=info"))
         }))
         .init();
 
@@ -98,6 +98,11 @@ async fn serve(path: &std::path::Path) -> Result<()> {
 
     let listener = tokio::net::TcpListener::bind(&cfg.bind).await.with_context(|| format!("listening on {}", cfg.bind))?;
     tracing::info!(version = env!("CARGO_PKG_VERSION"), bind = %cfg.bind, tokens = cfg.tokens.len(), dropped, "directory started");
+    if !cfg.metrics_bind.is_empty() {
+        let metrics = tokio::net::TcpListener::bind(&cfg.metrics_bind).await.with_context(|| format!("listening for metrics on {}", cfg.metrics_bind))?;
+        tracing::info!(bind = %cfg.metrics_bind, "serving metrics for Prometheus at /metrics");
+        tokio::spawn(ipx_directory::metrics::serve(metrics));
+    }
     tokio::spawn(ipx_directory::refresher(dir.clone()));
     tokio::spawn(ipx_directory::reloader(dir.clone(), path.to_owned()));
     axum::serve(listener, ipx_directory::router(dir))

crates/directory/src/metrics.rs

@@ -0,0 +1,67 @@
+//! Memory, for ipx and the directory both (#224), and the directory's own /metrics.
+//!
+//! mimalloc is the allocator of every binary that builds this crate in, ipx included: with
+//! glibc's, ipx kept the 500-850 MB a night of scans had needed until it restarted, and the
+//! directory sat at 1.25 GiB doing nothing, each thread's arena holding what it had freed.
+//! mimalloc gives freed pages back, so resident memory follows what is in use: a spike that
+//! does not come down again is a leak. Not jemalloc, whose upstream was archived in 2025.
+
+use std::sync::atomic::Ordering::Relaxed;
+
+#[global_allocator]
+static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+/// The process's resident memory in bytes and its open files, from /proc; None elsewhere.
+pub fn process() -> (Option<f64>, Option<f64>) {
+    let rss = std::fs::read_to_string("/proc/self/status").ok().and_then(|s| {
+        let kb = s.lines().find_map(|l| l.strip_prefix("VmRSS:"))?.trim().strip_suffix("kB")?.trim().parse::<f64>().ok()?;
+        Some(kb * 1024.0)
+    });
+    let fds = std::fs::read_dir("/proc/self/fd").ok().map(|d| d.count() as f64);
+    (rss, fds)
+}
+
+/// The directory's /metrics, in Prometheus's text format: memory, open files, and the feeds it
+/// has fetched. Nothing from the database, so a scrape costs nothing and answers while it is down.
+fn render(started: std::time::Instant) -> String {
+    let mut out = String::new();
+    let mut gauge = |name: &str, kind: &str, help: &str, labels: &str, v: f64| {
+        out.push_str(&format!("# HELP {name} {help}\n# TYPE {name} {kind}\n{name}{labels} {v}\n"));
+    };
+    let (rss, fds) = process();
+    if let Some(v) = rss {
+        gauge("directory_process_resident_bytes", "gauge", "Memory the process holds.", "", v);
+    }
+    if let Some(v) = fds {
+        gauge("directory_process_open_fds", "gauge", "Files and sockets the process has open.", "", v);
+    }
+    gauge("directory_feed_fetches_total", "counter", "Feeds fetched, answered 304 included.", "", ipx_feed::FETCHES.load(Relaxed) as f64);
+    gauge("directory_feed_bytes_total", "counter", "Bytes of feed bodies fetched.", "", ipx_feed::FETCHED_BYTES.load(Relaxed) as f64);
+    gauge("directory_uptime_seconds", "gauge", "Seconds since the directory started.", "", started.elapsed().as_secs() as f64);
+    gauge("directory_build_info", "gauge", "The version running, as a label.", &format!("{{version=\"{}\"}}", env!("CARGO_PKG_VERSION")), 1.0);
+    out
+}
+
+/// Serves /metrics on a listener of its own: the public port is behind the tunnel with no
+/// Access in front, and these are for the LAN.
+pub async fn serve(listener: tokio::net::TcpListener) {
+    let started = std::time::Instant::now();
+    let app = axum::Router::new().route(
+        "/metrics",
+        axum::routing::get(move || async move {
+            ([(axum::http::header::CONTENT_TYPE, "text/plain; version=0.0.4")], render(started))
+        }),
+    );
+    if let Err(e) = axum::serve(listener, app).await {
+        tracing::error!(error = %e, "metrics stopped");
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn metrics_say_memory_and_feeds() {
+        let text = super::render(std::time::Instant::now());
+        assert!(text.contains("directory_process_resident_bytes ") && text.contains("directory_feed_bytes_total "), "{text}");
+    }
+}

crates/feed/src/lib.rs

@@ -61,13 +61,19 @@ pub enum Fetched {
 /// downloads, which are not, have no such limit.
 pub const FEED_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
 
+/// Feeds fetched and the bytes of their bodies, since the process started, for /metrics to set
+/// beside memory (#224): a body is read whole, so a night of large feeds is a night of large
+/// allocations.
+pub static FETCHES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+pub static FETCHED_BYTES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
 /// Conditional GET, following redirects itself: `client` must follow none (`feed_client`), so
 /// each hop is seen. The second value is where the feed now is when every hop said so for good
 /// (301 or 308): a publisher that moved its feed, which the catalogue should follow rather
 /// than be redirected on every read. A temporary redirect (302, 307) moves nothing.
 /// `system-proxy` reads http_proxy/https_proxy.
 /// `auth` is a username and password, sent only to the feed's own host.
-#[tracing::instrument(skip_all, fields(url = %feed_url))]
+#[tracing::instrument(skip_all, fields(url = %feed_url, http.response.body.size))]
 pub async fn fetch(
     client: &reqwest::Client,
     feed_url: &str,
@@ -106,6 +112,7 @@ pub async fn fetch(
             continue;
         }
         let moved = (permanent && url != start).then(|| url.to_string());
+        FETCHES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         if status == StatusCode::NOT_MODIFIED {
             return Ok((Fetched::NotModified, moved));
         }
@@ -119,6 +126,8 @@ pub async fn fetch(
         let etag = header(ETAG);
         let last_modified = header(LAST_MODIFIED);
         let bytes = resp.bytes().await.context("reading body")?.to_vec();
+        tracing::Span::current().record("http.response.body.size", bytes.len());
+        FETCHED_BYTES.fetch_add(bytes.len() as u64, std::sync::atomic::Ordering::Relaxed);
         return Ok((Fetched::Body { bytes, etag, last_modified }, moved));
     }
     // Worded as reqwest worded it, which failure_kind reads as a redirect loop.
@@ -364,6 +373,7 @@ fn parse_patreon_shows(json: &[u8]) -> Result<(Option<String>, Vec<(String, Stri
 }
 
 /// RSS first, then Atom -- the same split the original made on `parsedFeed.version`.
+#[tracing::instrument(skip_all, fields(bytes = bytes.len()))]
 pub fn parse(bytes: &[u8]) -> Result<ParsedFeed> {
     match rss::Channel::read_from(bytes) {
         Ok(ch) => Ok(from_rss(ch, bytes)),

crates/ipx/CHANGELOG.md

@@ -12,8 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `/metrics` counts the feeds fetched and the bytes of their bodies (`ipx_feed_fetches_total`, `ipx_feed_bytes_total`), and a feed's trace says how large its body was (#224).
+
 ### Changed
 
+- ipx uses mimalloc instead of glibc's allocator, so memory a night of scans needed goes back to the system instead of staying held until a restart (#224).
 - `docker-compose.yml` has no healthchecks: Docker runs them with exec, which can fail on the host while ipx is well. The library's counts in `/metrics` are kept every 30 s by the daemon itself (#218).
 
 ## [1.0.0] - 2026-10-12

crates/ipx/src/main.rs

@@ -299,7 +299,7 @@ async fn main() -> Result<()> {
                 use opentelemetry::trace::TracerProvider;
                 tracing_opentelemetry::layer()
                     .with_tracer(p.tracer("ipx"))
-                    .with_filter(tracing_subscriber::EnvFilter::new("ipx=info"))
+                    .with_filter(tracing_subscriber::EnvFilter::new("ipx=info,ipx_feed=info"))
             }))
             .init();
     }

crates/ipx/src/metrics.rs

@@ -169,13 +169,15 @@ pub fn render(s: &Sources) -> String {
     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)]);
     gauge(&mut out, "ipx_runtime_queue", "Tasks waiting in the runtime's global queue.", &[("", rt.global_queue_depth() as f64)]);
-    let (rss, fds) = process();
+    let (rss, fds) = ipx_directory::metrics::process();
     if let Some(v) = rss {
         gauge(&mut out, "ipx_process_resident_bytes", "Memory the process holds.", &[("", v)]);
     }
     if let Some(v) = fds {
         gauge(&mut out, "ipx_process_open_fds", "Files and sockets the process has open.", &[("", v)]);
     }
+    counter(&mut out, "ipx_feed_fetches_total", "Feeds fetched, answered 304 included.", &[("", ipx_feed::FETCHES.load(Relaxed) as f64)]);
+    counter(&mut out, "ipx_feed_bytes_total", "Bytes of feed bodies fetched.", &[("", ipx_feed::FETCHED_BYTES.load(Relaxed) as f64)]);
     gauge(&mut out, "ipx_uptime_seconds", "Seconds since the daemon started.", &[("", s.started.elapsed().as_secs() as f64)]);
     let version = format!("version=\"{}\"", env!("CARGO_PKG_VERSION"));
     gauge(&mut out, "ipx_build_info", "The version running, as a label.", &[(&version, 1.0)]);
@@ -248,16 +250,6 @@ fn write(out: &mut String, kind: &str, name: &str, help: &str, samples: &[(&str,
     }
 }
 
-/// The process's resident memory in bytes and its open files, from /proc; None elsewhere.
-pub fn process() -> (Option<f64>, Option<f64>) {
-    let rss = std::fs::read_to_string("/proc/self/status").ok().and_then(|s| {
-        let kb = s.lines().find_map(|l| l.strip_prefix("VmRSS:"))?.trim().strip_suffix("kB")?.trim().parse::<f64>().ok()?;
-        Some(kb * 1024.0)
-    });
-    let fds = std::fs::read_dir("/proc/self/fd").ok().map(|d| d.count() as f64);
-    (rss, fds)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -299,7 +291,7 @@ mod tests {
 
     #[test]
     fn the_process_reads_its_memory_and_files_on_linux() {
-        let (rss, fds) = process();
+        let (rss, fds) = ipx_directory::metrics::process();
         assert!(rss.unwrap() > 0.0 && fds.unwrap() > 0.0);
     }
 }