diff --git a/crates/directory/src/apple.rs b/crates/directory/src/apple.rs index 7318e4a..ca90a19 100644 --- a/crates/directory/src/apple.rs +++ b/crates/directory/src/apple.rs @@ -96,8 +96,11 @@ async fn sync(dir: &Shared, client: &reqwest::Client, country: &str) -> Result<( } let result = dir.store.snapshot(INSTANCE, &entries).await?; tracing::info!( - shows = ids.len(), accepted = result.accepted, refused_private = result.refused_private, - refused_blocked = result.refused_blocked, refused_invalid = result.refused_invalid, + shows = ids.len(), + accepted = result.accepted, + refused_private = result.refused_private, + refused_blocked = result.refused_blocked, + refused_invalid = result.refused_invalid, "Apple Podcasts synced" ); // Fetched now, not when the refresher next looks. @@ -128,7 +131,11 @@ struct Genre { fn categories(g: &Genres) -> BTreeMap> { g.0.get("26") .map(|podcasts| { - podcasts.subgenres.values().map(|c| (c.name.clone(), c.subgenres.values().map(|s| s.name.clone()).collect())).collect() + podcasts + .subgenres + .values() + .map(|c| (c.name.clone(), c.subgenres.values().map(|s| s.name.clone()).collect())) + .collect() }) .unwrap_or_default() } @@ -253,7 +260,8 @@ mod tests { ) .unwrap(); assert_eq!(chart_ids(&chart), ["1200361736", "1028908750"]); - let one: Chart = serde_json::from_str(r#"{"feed": {"entry": {"id": {"attributes": {"im:id": "7"}}}}}"#).unwrap(); + let one: Chart = + serde_json::from_str(r#"{"feed": {"entry": {"id": {"attributes": {"im:id": "7"}}}}}"#).unwrap(); assert_eq!(chart_ids(&one), ["7"], "a chart of one"); let empty: Chart = serde_json::from_str(r#"{"feed": {}}"#).unwrap(); assert!(chart_ids(&empty).is_empty()); diff --git a/crates/directory/src/config.rs b/crates/directory/src/config.rs index d2aaed1..457526f 100644 --- a/crates/directory/src/config.rs +++ b/crates/directory/src/config.rs @@ -162,7 +162,10 @@ mod tests { let cfg = Config::parse(GOOD).unwrap(); assert_eq!(cfg.bind, "0.0.0.0:8100"); let t = cfg.tokens().unwrap(); - assert_eq!((t[0].name.as_str(), t[0].access, t[0].sha256.as_str()), ("tower", Access::Rwa, "a".repeat(64).as_str())); + assert_eq!( + (t[0].name.as_str(), t[0].access, t[0].sha256.as_str()), + ("tower", Access::Rwa, "a".repeat(64).as_str()) + ); assert_eq!(t[1].access, Access::Ro); assert!(Access::Ro < Access::Rw && Access::Rw < Access::Rwa); } @@ -171,7 +174,9 @@ mod tests { fn a_missing_or_misspelled_access_a_reused_name_or_a_bad_hash_is_refused() { let h = "a".repeat(64); assert!(Config::parse(&format!("[[tokens]]\nname = \"a\"\ntoken_sha256 = \"{h}\"")).is_err()); - assert!(Config::parse(&format!("[[tokens]]\nname = \"a\"\naccess = \"admin\"\ntoken_sha256 = \"{h}\"")).is_err()); + assert!( + Config::parse(&format!("[[tokens]]\nname = \"a\"\naccess = \"admin\"\ntoken_sha256 = \"{h}\"")).is_err() + ); assert!(Config::parse(&GOOD.replace("\"friend\"", "\"tower\"")).is_err()); assert!(Config::parse(&GOOD.replace(&"b".repeat(64), "xyz")).unwrap().tokens().is_err()); assert!(Config::parse(&GOOD.replace(&"b".repeat(64), &h)).unwrap().tokens().is_err()); diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 2ecf150..3599c0b 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -153,16 +153,16 @@ impl Directory { fn caller(&self, presented: &str) -> Option { let h = config::hash(presented); // Every token compared, whether or not one matched, so timing says nothing either way. - self.tokens - .read() - .unwrap() - .iter() - .fold(None, |found, t| if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found }) + self.tokens.read().unwrap().iter().fold(None, |found, t| { + if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found } + }) } /// Fetches one feed and stores what it found. pub async fn refresh(&self, due: &store::Due) -> anyhow::Result<()> { - let got = match ipx_feed::fetch(&self.client, &due.url, None, due.etag.as_deref(), due.last_modified.as_deref()).await { + let got = match ipx_feed::fetch(&self.client, &due.url, None, due.etag.as_deref(), due.last_modified.as_deref()) + .await + { Ok((ipx_feed::Fetched::NotModified, _)) => Refreshed::NotModified, Ok((ipx_feed::Fetched::Body { bytes, etag, last_modified }, _)) => { if ipx_feed::is_opml(&bytes) { @@ -242,10 +242,14 @@ pub async fn reloader(dir: Shared, path: std::path::PathBuf) { let n = tokens.len(); match dir.set_tokens(tokens).await { Ok(dropped) => tracing::info!(tokens = n, dropped, "config read again"), - Err(e) => tracing::warn!(error = %format!("{e:#}"), "config read again, but old snapshots could not be dropped"), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "config read again, but old snapshots could not be dropped") + } } } - Err(e) => tracing::warn!(error = %format!("{e:#}"), "config changed but does not load; keeping the tokens as they were"), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "config changed but does not load; keeping the tokens as they were") + } } } } @@ -342,7 +346,8 @@ async fn authenticate(State(dir): State, mut req: Request, next: Next) - .map(str::trim) .filter(|t| !t.is_empty()); let Some(me) = presented.and_then(|t| dir.caller(t)) else { - return ApiError(StatusCode::UNAUTHORIZED, "a valid token is needed: Authorization: Bearer ".into()).into_response(); + return ApiError(StatusCode::UNAUTHORIZED, "a valid token is needed: Authorization: Bearer ".into()) + .into_response(); }; let who = Who(me.name.clone()); req.extensions_mut().insert(me); @@ -387,7 +392,11 @@ async fn popular(State(dir): State, Query(q): Query) -> Result, Path(id): Path) -> Result, ApiError> { - dir.store.feed(&id).await?.map(Json).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory"))) + dir.store + .feed(&id) + .await? + .map(Json) + .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory"))) } async fn snapshot( @@ -408,7 +417,11 @@ async fn snapshot( Ok(Json(r)) } -async fn remove(State(dir): State, Extension(me): Extension, Path(id): Path) -> Result { +async fn remove( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result { need(&me, Access::Rwa)?; dir.store.block(&id).await?; tracing::info!(feed = %id, by = %me.name, "feed removed and blocked"); @@ -420,7 +433,11 @@ async fn blocked(State(dir): State, Extension(me): Extension) -> Res Ok(Json(dir.store.blocked().await?)) } -async fn unblock(State(dir): State, Extension(me): Extension, Path(id): Path) -> Result { +async fn unblock( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result { need(&me, Access::Rwa)?; if !dir.store.unblock(&id).await? { return Err(ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not blocked"))); @@ -434,16 +451,28 @@ async fn instances(State(dir): State, Extension(me): Extension) -> R Ok(Json(dir.store.instances().await?)) } -async fn drop_instance(State(dir): State, Extension(me): Extension, Path(name): Path) -> Result { +async fn drop_instance( + State(dir): State, + Extension(me): Extension, + Path(name): Path, +) -> Result { need(&me, Access::Rwa)?; let n = dir.store.drop_instance(&name).await?; tracing::info!(instance = %name, feeds = n, by = %me.name, "snapshot dropped"); Ok(StatusCode::NO_CONTENT) } -async fn refresh(State(dir): State, Extension(me): Extension, Path(id): Path) -> Result, ApiError> { +async fn refresh( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result, ApiError> { need(&me, Access::Rwa)?; - let due = dir.store.due_one(&id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))?; + let due = dir + .store + .due_one(&id) + .await? + .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))?; dir.refresh(&due).await?; feed(State(dir), Path(id)).await } @@ -467,7 +496,13 @@ mod tests { (router(dir.clone()), dir) } - async fn call(app: &Router, method: &str, path: &str, token: Option<&str>, body: Option) -> (u16, serde_json::Value) { + async fn call( + app: &Router, + method: &str, + path: &str, + token: Option<&str>, + body: Option, + ) -> (u16, serde_json::Value) { let mut req = axum::http::Request::builder().method(method).uri(path); if let Some(t) = token { req = req.header("authorization", format!("Bearer secret-{t}")); @@ -501,7 +536,8 @@ mod tests { .unwrap(); let resp = app.clone().oneshot(req).await.unwrap(); assert_eq!(resp.status(), 200); - let got: serde_json::Value = serde_json::from_slice(&axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap()).unwrap(); + let got: serde_json::Value = + serde_json::from_slice(&axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap()).unwrap(); assert_eq!(got["accepted"], 1); } @@ -510,11 +546,8 @@ mod tests { let (app, _) = app().await; for (asked, sent) in [("br", Some("br")), ("zstd", Some("zstd")), ("gzip", Some("gzip")), ("identity", None)] { // A page, not an empty feed list: under 32 bytes is sent as it is. - let req = axum::http::Request::builder() - .uri("/") - .header("accept-encoding", asked) - .body(Body::empty()) - .unwrap(); + let req = + axum::http::Request::builder().uri("/").header("accept-encoding", asked).body(Body::empty()).unwrap(); let resp = app.clone().oneshot(req).await.unwrap(); assert_eq!(resp.status(), 200); let got = resp.headers().get("content-encoding").map(|v| v.to_str().unwrap().to_owned()); @@ -583,7 +616,11 @@ mod tests { assert_eq!(call(&app, "GET", "/v1/feeds/nope", Some("ro"), None).await.0, 404); assert_eq!(call(&app, "GET", "/v1/popular?limit=5", Some("ro"), None).await.1[0]["id"], id.as_str()); - for (m, p) in [("DELETE", format!("/v1/feeds/{id}")), ("GET", "/v1/blocked".into()), ("DELETE", "/v1/instances/rw-token".into())] { + for (m, p) in [ + ("DELETE", format!("/v1/feeds/{id}")), + ("GET", "/v1/blocked".into()), + ("DELETE", "/v1/instances/rw-token".into()), + ] { assert_eq!(call(&app, m, &p, Some("rw"), None).await.0, 403, "{m} {p}"); } assert_eq!(call(&app, "DELETE", &format!("/v1/feeds/{id}"), Some("rwa"), None).await.0, 204); diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 910e26c..1d8fc09 100644 --- a/crates/directory/src/main.rs +++ b/crates/directory/src/main.rs @@ -74,15 +74,24 @@ async fn main() -> Result<()> { async fn serve(path: &std::path::Path) -> Result<()> { // One JSON object a line for Loki when DIRECTORY_LOG_FORMAT=json, as ipx does; text otherwise. let json = std::env::var("DIRECTORY_LOG_FORMAT").is_ok_and(|f| f.eq_ignore_ascii_case("json")); - let filter = || tracing_subscriber::EnvFilter::try_from_env("DIRECTORY_LOG").unwrap_or_else(|_| "ipx_directory=info".into()); + let filter = + || tracing_subscriber::EnvFilter::try_from_env("DIRECTORY_LOG").unwrap_or_else(|_| "ipx_directory=info".into()); // Traces to Tempo when OTEL_EXPORTER_OTLP_ENDPOINT is set, joining ipx's (#213). let otel = ipx_directory::otel::provider("ipx-directory")?; tracing_subscriber::registry() .with((!json).then(|| tracing_subscriber::fmt::layer().with_writer(std::io::stderr).with_filter(filter()))) - .with(json.then(|| tracing_subscriber::fmt::layer().json().flatten_event(true).with_writer(std::io::stderr).with_filter(filter()))) + .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,ipx_feed=info")) + tracing_opentelemetry::layer() + .with_tracer(p.tracer("ipx-directory")) + .with_filter(tracing_subscriber::EnvFilter::new("ipx_directory=info,ipx_feed=info")) })) .init(); @@ -96,10 +105,13 @@ async fn serve(path: &std::path::Path) -> Result<()> { let dir = Directory::new(store, vec![])?; let dropped = dir.set_tokens(tokens).await?; - let listener = tokio::net::TcpListener::bind(&cfg.bind).await.with_context(|| format!("listening on {}", cfg.bind))?; + 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))?; + 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)); } @@ -112,7 +124,8 @@ async fn serve(path: &std::path::Path) -> Result<()> { tokio::spawn(ipx_directory::reloader(dir.clone(), path.to_owned())); axum::serve(listener, ipx_directory::router(dir)) .with_graceful_shutdown(async { - let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("SIGTERM handler"); + let mut term = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("SIGTERM handler"); tokio::select! { _ = term.recv() => {} _ = tokio::signal::ctrl_c() => {} diff --git a/crates/directory/src/metrics.rs b/crates/directory/src/metrics.rs index f26d045..90a0cf0 100644 --- a/crates/directory/src/metrics.rs +++ b/crates/directory/src/metrics.rs @@ -14,7 +14,8 @@ 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, Option) { 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::().ok()?; + let kb = + s.lines().find_map(|l| l.strip_prefix("VmRSS:"))?.trim().strip_suffix("kB")?.trim().parse::().ok()?; Some(kb * 1024.0) }); let fds = std::fs::read_dir("/proc/self/fd").ok().map(|d| d.count() as f64); @@ -35,10 +36,34 @@ fn render(started: std::time::Instant) -> String { 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); + 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 } @@ -46,14 +71,15 @@ fn render(started: std::time::Instant) -> String { /// 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)) - }), - ) - // Prometheus asks for gzip (#253). - .layer(tower_http::compression::CompressionLayer::new().gzip(true)); + 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)) + }), + ) + // Prometheus asks for gzip (#253). + .layer(tower_http::compression::CompressionLayer::new().gzip(true)); if let Err(e) = axum::serve(listener, app).await { tracing::error!(error = %e, "metrics stopped"); } @@ -64,6 +90,9 @@ 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}"); + assert!( + text.contains("directory_process_resident_bytes ") && text.contains("directory_feed_bytes_total "), + "{text}" + ); } } diff --git a/crates/directory/src/otel.rs b/crates/directory/src/otel.rs index deb31cb..bdf5fd1 100644 --- a/crates/directory/src/otel.rs +++ b/crates/directory/src/otel.rs @@ -70,13 +70,21 @@ mod tests { assert_eq!(cx.span().span_context().trace_id().to_string(), "4bf92f3577b34da6a3ce929d0e0e4736"); assert!(cx.span().span_context().is_remote()); // Anything else from the browser starts a trace of its own. - for bad in ["", "00-xyz-00f067aa0ba902b7-01", "01-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", - "00-00000000000000000000000000000000-00f067aa0ba902b7-01", "00-4bf92f3577b34da6a3ce929d0e0e4736"] { + for bad in [ + "", + "00-xyz-00f067aa0ba902b7-01", + "01-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + "00-00000000000000000000000000000000-00f067aa0ba902b7-01", + "00-4bf92f3577b34da6a3ce929d0e0e4736", + ] { assert!(with(bad).is_none(), "{bad}"); } assert!(remote_parent(&axum::http::HeaderMap::new(), None).is_none()); // From the query, where a beacon or Safari's player cannot send a header (#192). - let q = remote_parent(&axum::http::HeaderMap::new(), Some("x=1&traceparent=00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")); + let q = remote_parent( + &axum::http::HeaderMap::new(), + Some("x=1&traceparent=00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), + ); assert_eq!(q.unwrap().span().span_context().trace_id().to_string(), "4bf92f3577b34da6a3ce929d0e0e4736"); assert!(remote_parent(&axum::http::HeaderMap::new(), Some("traceparent=nonsense")).is_none()); } diff --git a/crates/directory/src/site.rs b/crates/directory/src/site.rs index 4f1d5da..9b0926d 100644 --- a/crates/directory/src/site.rs +++ b/crates/directory/src/site.rs @@ -9,11 +9,11 @@ use std::collections::BTreeMap; use std::fmt::Write; +use axum::Router; use axum::extract::{Path, Query, State}; use axum::http::{HeaderValue, StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::get; -use axum::Router; use crate::{Listing, Shared}; @@ -63,7 +63,9 @@ async fn home(State(dir): State) -> Response { Err(e) => return failed(e), }; let mut top = all.clone(); - top.sort_by(|a, b| b.subscribers.cmp(&a.subscribers).then_with(|| name(a).to_lowercase().cmp(&name(b).to_lowercase()))); + top.sort_by(|a, b| { + b.subscribers.cmp(&a.subscribers).then_with(|| name(a).to_lowercase().cmp(&name(b).to_lowercase())) + }); top.truncate(10); let mut cats: BTreeMap<&str, Vec<&Listing>> = BTreeMap::new(); for l in &all { @@ -75,12 +77,25 @@ async fn home(State(dir): State) -> Response { } else { b.push_str("

Most subscribed

    "); for (i, l) in top.iter().enumerate() { - let _ = write!(b, "
  1. {}{}{}{}
    2. ", - esc(&l.id), i + 1, art(l, "art sm"), esc(name(l)), subscribers(l.subscribers)); + let _ = write!( + b, + "
  2. {}{}{}{}
    3. ", + esc(&l.id), + i + 1, + art(l, "art sm"), + esc(name(l)), + subscribers(l.subscribers) + ); } b.push_str("

Categories

"); for (c, ls) in &cats { - let _ = write!(b, "{}{}", seg(c), esc(c), ls.len()); + let _ = write!( + b, + "{}{}", + seg(c), + esc(c), + ls.len() + ); } let _ = write!(b, "

All {} feeds, A to Z

", all.len()); } @@ -115,10 +130,11 @@ async fn listed(dir: &Shared, title: &str, here: &str, f: &Filter, pick: impl Fn if ls.is_empty() && here.starts_with("/category/") { return not_found("That category has nothing in it."); } - let subs: BTreeMap = ls.iter().filter_map(|l| l.subcategory.clone()).fold(BTreeMap::new(), |mut m, s| { - *m.entry(s).or_default() += 1; - m - }); + let subs: BTreeMap = + ls.iter().filter_map(|l| l.subcategory.clone()).fold(BTreeMap::new(), |mut m, s| { + *m.entry(s).or_default() += 1; + m + }); ls.retain(|l| f.keep(l) && f.sub.as_ref().is_none_or(|s| l.subcategory.as_ref() == Some(s))); ls.sort_by_key(|l| name(l).to_lowercase()); @@ -138,18 +154,38 @@ async fn listed(dir: &Shared, title: &str, here: &str, f: &Filter, pick: impl Fn if q.is_empty() { here.to_owned() } else { format!("{here}?{q}") } }; let mut b = String::new(); - let _ = write!(b, "

Directory

{}