diff --git a/crates/directory/src/apple.rs b/crates/directory/src/apple.rs index ca90a19..bbe5660 100644 --- a/crates/directory/src/apple.rs +++ b/crates/directory/src/apple.rs @@ -13,6 +13,7 @@ use serde::Deserialize; use tracing::Instrument; use crate::{Shared, SnapshotEntry}; +use std::collections::{BTreeSet, HashSet}; /// What the charts are filed under, as a server's snapshot is under its token's name. pub const INSTANCE: &str = "apple-podcasts"; @@ -83,7 +84,7 @@ async fn sync(dir: &Shared, client: &reqwest::Client, country: &str) -> Result<( } tokio::time::sleep(PAUSE).await; } - let mut seen = std::collections::HashSet::new(); + let mut seen = HashSet::new(); ids.retain(|id| seen.insert(id.clone())); anyhow::ensure!(!ids.is_empty(), "no chart had a show in it"); @@ -150,9 +151,8 @@ fn differ(theirs: &BTreeMap>, ours: &[(&str, &[&str])]) -> ( let flat = |c: &str, subs: &mut dyn Iterator| { std::iter::once(c.to_string()).chain(subs.map(|s| format!("{c} / {s}"))).collect::>() }; - let theirs: std::collections::BTreeSet = - theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect(); - let ours: std::collections::BTreeSet = + let theirs: BTreeSet = theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect(); + let ours: BTreeSet = ours.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().map(|s| s.to_string()))).collect(); (theirs.difference(&ours).cloned().collect(), ours.difference(&theirs).cloned().collect()) } diff --git a/crates/directory/src/config.rs b/crates/directory/src/config.rs index 81bc6a8..d96e072 100644 --- a/crates/directory/src/config.rs +++ b/crates/directory/src/config.rs @@ -4,6 +4,8 @@ use anyhow::{Context, Result, bail}; use serde::Deserialize; +use std::collections::HashSet; +use std::fs::File; use std::path::Path; /// `directory.toml`. See docs/directory.md. @@ -87,7 +89,7 @@ impl Config { /// instances' snapshots together. pub fn parse(text: &str) -> Result { let cfg: Config = toml::from_str(text)?; - let mut seen = std::collections::HashSet::new(); + let mut seen = HashSet::new(); for t in &cfg.tokens { if t.name.trim().is_empty() { bail!("a token has an empty name"); @@ -111,7 +113,7 @@ impl Config { /// The tokens as the server checks them. A hash that is not 64 hex digits, or the same /// hash under two names, is an error: which name a request is would be ambiguous. pub fn tokens(&self) -> Result> { - let mut seen = std::collections::HashSet::new(); + let mut seen = HashSet::new(); self.tokens .iter() .map(|t| { @@ -148,9 +150,7 @@ pub fn new_token() -> String { use std::io::Read; let mut bytes = [0u8; 32]; // Falling back to the clock would make a guessable token. Better to fail. - std::fs::File::open("/dev/urandom") - .and_then(|mut f| f.read_exact(&mut bytes)) - .expect("no source of randomness for a token"); + File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut bytes)).expect("no source of randomness for a token"); format!("ipxd_{}", bytes.iter().map(|b| format!("{b:02x}")).collect::()) } diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 2670835..032282b 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -26,7 +26,13 @@ use axum::{Extension, Json, Router}; use serde::{Deserialize, Serialize}; use config::{Access, Token}; +use reqwest::redirect::Policy; +use std::path::PathBuf; use store::{Refreshed, Store}; +use tokio::sync::Notify; +use tower_http::compression::CompressionLayer; +use tower_http::decompression::RequestDecompressionLayer; +use tracing::Instrument; /// A feed as the directory lists it. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -134,7 +140,7 @@ pub struct Directory { tokens: RwLock>, client: reqwest::Client, /// Wakes the refresher, so a feed a snapshot brings in is fetched now, not in a minute. - wake: tokio::sync::Notify, + wake: Notify, /// The public pages' copy of the list, and when it was read. pages_list: tokio::sync::Mutex>)>>, } @@ -155,9 +161,9 @@ impl Directory { ) -> Result { // No redirects: ipx_feed::fetch follows them itself, so it sees each hop. let client = reqwest::Client::builder() - .connect_timeout(std::time::Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)) .read_timeout(ipx_feed::FEED_SILENCE) - .redirect(reqwest::redirect::Policy::none()) + .redirect(Policy::none()) .user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION"))) // The feeds are named by whoever holds a write token, and redirect where their // publishers like: nothing on this network is fetched unless listed (#329). @@ -167,7 +173,7 @@ impl Directory { store, tokens: RwLock::new(tokens), client, - wake: tokio::sync::Notify::new(), + wake: Notify::new(), pages_list: Default::default(), })) } @@ -239,14 +245,14 @@ pub async fn refresher(dir: Shared) { if !due.is_empty() { // A round is a trace, each feed's check a span in it (#213). let round = tracing::info_span!("refresh", feeds = due.len()); - tracing::Instrument::instrument( + Instrument::instrument( async { // A few at a time: in turn, two publishers timing out held a round // for a minute and more (#227). futures_util::StreamExt::for_each_concurrent(futures_util::stream::iter(&due), 5, |d| { let check = tracing::info_span!("feed", feed = %d.id); async { - if let Err(e) = tracing::Instrument::instrument(dir.refresh(d), check).await { + if let Err(e) = Instrument::instrument(dir.refresh(d), check).await { tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check"); } } @@ -272,7 +278,7 @@ pub async fn refresher(dir: Shared) { /// Reads the config file again whenever it changes, checked each minute. A file that fails to /// load leaves the tokens as they were, so a typo locks nobody out. -pub async fn reloader(dir: Shared, path: std::path::PathBuf) { +pub async fn reloader(dir: Shared, path: PathBuf) { let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok(); let mut last = mtime(&path); loop { @@ -321,9 +327,9 @@ pub fn router(dir: Shared) -> Router { // Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of // JSON, most of a Directory page's time (#253). Inside the access log, which records what // was sent. - .layer(tower_http::compression::CompressionLayer::new().br(true).zstd(true).gzip(true)) + .layer(CompressionLayer::new().br(true).zstd(true).gzip(true)) // And takes a request's body gzipped: ipx sends its snapshot so, about 100 KB as JSON. - .layer(tower_http::decompression::RequestDecompressionLayer::new().gzip(true)) + .layer(RequestDecompressionLayer::new().gzip(true)) .layer(axum::middleware::from_fn(access_log)) .with_state(dir) } @@ -368,7 +374,7 @@ async fn access_log(req: Request, next: Next) -> Response { use tracing_opentelemetry::OpenTelemetrySpanExt; let _ = span.set_parent(parent); } - let resp = tracing::Instrument::instrument(next.run(req), span.clone()).await; + let resp = Instrument::instrument(next.run(req), span.clone()).await; span.record("http.response.status_code", resp.status().as_u16()); let _in = span.enter(); let who = resp.extensions().get::().map_or("", |w| w.0.as_str()).to_owned(); diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 5233dfd..07668c0 100644 --- a/crates/directory/src/main.rs +++ b/crates/directory/src/main.rs @@ -4,6 +4,11 @@ use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; use ipx_directory::store::Store; use ipx_directory::{Directory, config}; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::signal::unix::SignalKind; +use tracing_subscriber::EnvFilter; use tracing_subscriber::prelude::*; #[derive(Parser)] @@ -12,7 +17,7 @@ struct Cli { /// The config file: where to listen, the database, and the tokens it accepts. Otherwise /// DIRECTORY_CONFIG, or else /config/directory.toml. #[arg(long)] - config: Option, + config: Option, #[command(subcommand)] cmd: Option, } @@ -62,7 +67,7 @@ async fn main() -> Result<()> { let port = cfg.bind.rsplit(':').next().unwrap_or("8100"); let resp = reqwest::Client::new() .get(format!("http://127.0.0.1:{port}/healthz")) - .timeout(std::time::Duration::from_secs(4)) + .timeout(Duration::from_secs(4)) .send() .await .context("the directory does not answer")?; @@ -72,11 +77,10 @@ async fn main() -> Result<()> { } } -async fn serve(path: &std::path::Path) -> Result<()> { +async fn serve(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 = || 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() @@ -92,7 +96,7 @@ async fn serve(path: &std::path::Path) -> Result<()> { 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")) + .with_filter(EnvFilter::new("ipx_directory=info,ipx_feed=info")) })) .init(); @@ -106,11 +110,10 @@ async fn serve(path: &std::path::Path) -> Result<()> { let dir = Directory::new(store, vec![], ipx_feed::guard::Allowed::parse(&cfg.fetch_internal)?)?; 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 = 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) + let metrics = 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"); @@ -125,8 +128,7 @@ 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(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 44d8999..c063808 100644 --- a/crates/directory/src/metrics.rs +++ b/crates/directory/src/metrics.rs @@ -6,7 +6,11 @@ //! 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 axum::Router; use std::sync::atomic::Ordering::Relaxed; +use std::time::Instant; +use tokio::net::TcpListener; +use tower_http::compression::CompressionLayer; #[global_allocator] static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; @@ -25,7 +29,7 @@ pub fn process() -> (Option, Option) { /// 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 { +fn render(started: 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")); @@ -70,9 +74,9 @@ fn render(started: std::time::Instant) -> String { /// 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() +pub async fn serve(listener: TcpListener) { + let started = Instant::now(); + let app = Router::new() .route( "/metrics", axum::routing::get(move || async move { @@ -80,7 +84,7 @@ pub async fn serve(listener: tokio::net::TcpListener) { }), ) // Prometheus asks for gzip (#253). - .layer(tower_http::compression::CompressionLayer::new().gzip(true)); + .layer(CompressionLayer::new().gzip(true)); if let Err(e) = axum::serve(listener, app).await { tracing::error!(error = %e, "metrics stopped"); } diff --git a/crates/directory/src/otel.rs b/crates/directory/src/otel.rs index e8a722f..207b4d8 100644 --- a/crates/directory/src/otel.rs +++ b/crates/directory/src/otel.rs @@ -2,6 +2,8 @@ //! caller and out to a callee, so one trace follows a request from the browser through ipx-web //! and ipx into the directory. +use axum::http::HeaderMap; + /// Traces over OTLP, to Tempo for one, when OTEL_EXPORTER_OTLP_ENDPOINT names a collector /// (`http://host:4318`: the exporter speaks OTLP over HTTP and adds `/v1/traces`). The exporter /// reads that and the other `OTEL_` variables itself. @@ -24,7 +26,7 @@ pub fn provider(service: &'static str) -> anyhow::Result) -> Option { +pub fn remote_parent(headers: &HeaderMap, query: Option<&str>) -> Option { use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId, TraceState}; let v = match headers.get("traceparent") { Some(h) => h.to_str().ok()?, diff --git a/crates/directory/src/site.rs b/crates/directory/src/site.rs index f0c072f..73e459d 100644 --- a/crates/directory/src/site.rs +++ b/crates/directory/src/site.rs @@ -16,6 +16,7 @@ use axum::response::{IntoResponse, Response}; use axum::routing::get; use crate::{Listing, Shared}; +use url::form_urlencoded::Serializer; pub fn router() -> Router { Router::new() @@ -140,7 +141,7 @@ async fn listed(dir: &Shared, title: &str, here: &str, f: &Filter, pick: impl Fn // The query a link keeps, with one of its keys changed. let link = |kind: Option<&str>, sub: Option<&str>| { - let mut q = url::form_urlencoded::Serializer::new(String::new()); + let mut q = Serializer::new(String::new()); if let Some(v) = f.q.as_deref() { q.append_pair("q", v); } diff --git a/crates/directory/src/store.rs b/crates/directory/src/store.rs index 127c313..6a7c472 100644 --- a/crates/directory/src/store.rs +++ b/crates/directory/src/store.rs @@ -6,6 +6,11 @@ use anyhow::Result; use sea_orm::{ConnectionTrait, DatabaseConnection, Statement, TransactionTrait, Value}; use crate::{Blocked, FeedPage, Instance, Item, Listing, SnapshotEntry, SnapshotResult}; +use sea_orm::{ConnectOptions, Database, QueryResult}; +use std::cmp::Reverse; +use std::collections::{BTreeMap, HashSet}; +use std::time::SystemTime; +use url::Url; /// A feed's items as its page shows them, and as many as ipx's own Directory page showed. pub const ITEMS: usize = 20; @@ -42,14 +47,14 @@ pub enum Refreshed { /// Unix seconds now. #[must_use] pub fn now() -> i64 { - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64) + SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64) } /// The address as every instance would write it: scheme and host lower-cased, a default port /// and a fragment dropped. None for anything that is not http(s). #[must_use] pub fn normalise(url: &str) -> Option { - let mut u = url::Url::parse(url.trim()).ok()?; + let mut u = Url::parse(url.trim()).ok()?; if !matches!(u.scheme(), "http" | "https") || u.host_str().is_none() { return None; } @@ -146,7 +151,7 @@ impl Store { let sep = if url.contains('?') { '&' } else { '?' }; url = format!("{url}{sep}options=-c%20client_min_messages%3Dwarning"); } - let mut opts = sea_orm::ConnectOptions::new(url.clone()); + let mut opts = ConnectOptions::new(url.clone()); opts.sqlx_logging(false); // An in-memory SQLite database is one per connection: a pool of them would be several. if url.contains(":memory:") { @@ -157,7 +162,7 @@ impl Store { if keep_open && url.starts_with("postgres") { opts.max_connections(10).min_connections(10); } - let db = sea_orm::Database::connect(opts).await?; + let db = Database::connect(opts).await?; for sql in SCHEMA { db.execute_unprepared(sql).await?; } @@ -177,7 +182,7 @@ impl Store { Ok(self.db.execute_raw(self.stmt(sql, values)).await?.rows_affected()) } - async fn rows(&self, sql: &str, values: Vec) -> Result> { + async fn rows(&self, sql: &str, values: Vec) -> Result> { Ok(self.db.query_all_raw(self.stmt(sql, values)).await?) } @@ -209,14 +214,14 @@ impl Store { /// blocked feed is left out and counted; the rest are added to the feeds, new ones to be /// fetched by the refresher. Feeds no instance sends any more go. pub async fn snapshot(&self, instance: &str, entries: &[SnapshotEntry]) -> Result { - let blocked: std::collections::HashSet = self + let blocked: HashSet = self .rows("SELECT id FROM blocked", vec![]) .await? .iter() .map(|r| r.try_get("", "id")) .collect::>()?; let mut result = SnapshotResult::default(); - let mut keep = std::collections::BTreeMap::new(); + let mut keep = BTreeMap::new(); for e in entries { let Some(url) = normalise(&e.url) else { result.refused_invalid += 1; @@ -307,7 +312,7 @@ impl Store { f.podcast, CAST(sum(s.subscribers) AS BIGINT) AS subscribers FROM feeds f JOIN snapshots s ON s.feed_id = f.id WHERE f.items IS NOT NULL"; - fn listing(r: &sea_orm::QueryResult) -> Result { + fn listing(r: &QueryResult) -> Result { Ok(Listing { id: r.try_get("", "id")?, url: r.try_get("", "url")?, @@ -483,7 +488,7 @@ impl Store { .flat_map(|e| &e.enclosures) .any(|e| e.mime.as_deref().is_some_and(|m| m.starts_with("audio/") || m.starts_with("video/"))); let mut entries = parsed.entries; - entries.sort_by_key(|e| std::cmp::Reverse(e.published)); + entries.sort_by_key(|e| Reverse(e.published)); // What an item is, never its files: their addresses are where a key would be. let items: Vec = entries .into_iter() diff --git a/crates/feed/src/guard.rs b/crates/feed/src/guard.rs index ee57f49..5f9c72e 100644 --- a/crates/feed/src/guard.rs +++ b/crates/feed/src/guard.rs @@ -14,6 +14,7 @@ use std::sync::Arc; use std::task::{Context, Poll}; use anyhow::{Context as _, Result}; +use http::Extensions; use hyper_util::client::legacy::connect::{Connection, HttpInfo}; type BoxError = Box; @@ -153,7 +154,7 @@ where let allowed = self.allowed.clone(); Box::pin(async move { let conn = connecting.await.map_err(Into::into)?; - let mut extras = http::Extensions::new(); + let mut extras = Extensions::new(); conn.connected().get_extras(&mut extras); // Refused when the peer is unknown too: a connection this cannot check is one it // cannot vouch for. diff --git a/crates/feed/src/lib.rs b/crates/feed/src/lib.rs index f491927..711ba34 100644 --- a/crates/feed/src/lib.rs +++ b/crates/feed/src/lib.rs @@ -4,8 +4,13 @@ pub mod guard; use anyhow::{Context, Result, anyhow}; +use chrono::{DateTime, NaiveDateTime}; use reqwest::StatusCode; +use reqwest::header::HeaderName; use reqwest::header::{ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED}; +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::AtomicU64; +use std::time::Duration; /// A feed as `parse` reads it, RSS or Atom alike: the channel's own details and its items. #[derive(Debug, Default)] @@ -73,20 +78,20 @@ pub enum Fetched { /// with no limit one hung server held every scan for as long as it did. Dreamwidth answered 504 /// after 60-67 s for a day, and each scan took 70-80 s instead of 15 (#108). A feed is small; /// downloads, which are not, have no such limit. -pub const FEED_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +pub const FEED_TIMEOUT: Duration = Duration::from_secs(30); /// The longest a feed's server may send nothing, for the clients that fetch feeds to set: one /// that took the connection and never answered ran the whole 30 s, every scan it was in (#230). /// No feed that answered took over 8 s in a day of traces; one that keeps sending, however /// slowly, still has until FEED_TIMEOUT. -pub const FEED_SILENCE: std::time::Duration = std::time::Duration::from_secs(15); +pub const FEED_SILENCE: Duration = Duration::from_secs(15); /// 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 FETCHES: AtomicU64 = AtomicU64::new(0); /// See `FETCHES`. -pub static FETCHED_BYTES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); +pub static FETCHED_BYTES: AtomicU64 = 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 @@ -204,8 +209,7 @@ pub async fn fetch( // The original surfaced 401/407 specially; the code is enough for a UI to switch on. return Err(anyhow!("HTTP {status}")); } - let header = - |h: reqwest::header::HeaderName| resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned); + let header = |h: HeaderName| resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned); let etag = header(ETAG); let last_modified = header(LAST_MODIFIED); let bytes = read_capped(resp).await.context("reading body")?; @@ -257,7 +261,7 @@ fn from_wp_posts(posts: &[serde_json::Value]) -> ParsedFeed { link: non_empty(p["link"].as_str()), published: p["date_gmt"] .as_str() - .and_then(|d| chrono::NaiveDateTime::parse_from_str(d, "%Y-%m-%dT%H:%M:%S").ok()) + .and_then(|d| NaiveDateTime::parse_from_str(d, "%Y-%m-%dT%H:%M:%S").ok()) .map(|d| d.and_utc().timestamp()), description: non_empty(p["content"]["rendered"].as_str()), image: non_empty(p["jetpack_featured_media_url"].as_str()), @@ -647,7 +651,7 @@ fn alternate_feed_link(bytes: &[u8]) -> Option { /// feed that moved (#102); cnn.com is one. pub async fn find_feed(client: &reqwest::Client, url: &str) -> Result { let read = |u: String| async move { - let resp = client.get(&u).timeout(std::time::Duration::from_secs(30)).send().await.context("connecting")?; + let resp = client.get(&u).timeout(Duration::from_secs(30)).send().await.context("connecting")?; anyhow::ensure!(resp.status().is_success(), "HTTP {}", resp.status()); let base = resp.url().clone(); anyhow::Ok((base, read_capped(resp).await.context("reading")?)) @@ -675,7 +679,7 @@ pub async fn find_feed(client: &reqwest::Client, url: &str) -> Result { /// `/favicon.ico`. None if neither is there. #[tracing::instrument(skip_all, fields(site = site))] pub async fn site_icon(client: &reqwest::Client, site: &str) -> Option { - let timeout = std::time::Duration::from_secs(20); + let timeout = Duration::from_secs(20); let resp = client.get(site).timeout(timeout).send().await.ok()?; // Relative to where the page ended up, not where it was asked for: a site that redirects // to /en/ would otherwise have its icon looked for in the wrong place. @@ -702,11 +706,7 @@ const MAX_ART_HOSTS: usize = 8; /// address it does not serve on https stays, for /api/art to fetch (#90). Four of the five /// hosts the catalogue had on http served the same image on https; The Secret Cabal's CDN /// presents another name's certificate (#110). -pub async fn prefer_https( - client: &reqwest::Client, - url: &str, - known: &mut std::collections::HashMap, -) -> String { +pub async fn prefer_https(client: &reqwest::Client, url: &str, known: &mut HashMap) -> String { let Some(rest) = url.strip_prefix("http://") else { return url.to_owned() }; let host = rest.split('/').next().unwrap_or("").to_owned(); let secure = format!("https://{rest}"); @@ -728,7 +728,7 @@ pub async fn prefer_https( /// Whether `url` answers with an image. A site with no favicon often answers 200 with its home /// page, which is not an icon. pub async fn is_image(client: &reqwest::Client, url: &str) -> bool { - let timeout = std::time::Duration::from_secs(20); + let timeout = Duration::from_secs(20); let Ok(resp) = client.get(url).timeout(timeout).send().await else { return false }; resp.status().is_success() && resp @@ -1037,7 +1037,7 @@ fn non_empty(s: Option<&str>) -> Option { /// that embeds the file it encloses lists the same file twice: Rands in Repose's "The Promotion /// Paradox" was downloaded twice and offered two play buttons for one mp3. The first stays. fn drop_player_repeats(encs: &mut Vec) { - let mut seen = std::collections::HashSet::new(); + let mut seen = HashSet::new(); encs.retain(|e| seen.insert(same_file_key(&e.url))); } @@ -1238,10 +1238,7 @@ fn parse_duration(s: &str) -> Option { /// RSS pubDate is RFC 2822; some feeds ship RFC 3339 instead. fn parse_date(s: &str) -> Option { let s = s.trim(); - chrono::DateTime::parse_from_rfc2822(s) - .or_else(|_| chrono::DateTime::parse_from_rfc3339(s)) - .ok() - .map(|d| d.timestamp()) + DateTime::parse_from_rfc2822(s).or_else(|_| DateTime::parse_from_rfc3339(s)).ok().map(|d| d.timestamp()) } /// Apple's podcast categories, each with its subcategories diff --git a/crates/ipx/src/art.rs b/crates/ipx/src/art.rs index 474d78b..01e1c11 100644 --- a/crates/ipx/src/art.rs +++ b/crates/ipx/src/art.rs @@ -2,8 +2,11 @@ //! of each publisher's server for every visit, and still there when a publisher is not. use std::collections::HashMap; +use std::collections::hash_map::DefaultHasher; +use std::fs::File; use std::path::{Path, PathBuf}; use std::sync::{LazyLock, Mutex}; +use std::time::SystemTime; use std::time::{Duration, Instant}; pub fn dir() -> PathBuf { @@ -16,7 +19,7 @@ pub fn dir() -> PathBuf { // one in 10^12. Move to a SHA if either ever matters. pub fn path(url: &str) -> PathBuf { use std::hash::{Hash, Hasher}; - let mut h = std::collections::hash_map::DefaultHasher::new(); + let mut h = DefaultHasher::new(); url.hash(&mut h); dir().join(format!("{:016x}", h.finish())) } @@ -25,8 +28,8 @@ pub fn path(url: &str) -> PathBuf { pub fn read(file: &Path) -> Option<(String, Vec)> { let kind = std::fs::read_to_string(file.with_extension("type")).ok()?; let body = std::fs::read(file).ok()?; - if let Ok(f) = std::fs::File::options().write(true).open(file) { - let _ = f.set_modified(std::time::SystemTime::now()); + if let Ok(f) = File::options().write(true).open(file) { + let _ = f.set_modified(SystemTime::now()); } Some((kind, body)) } @@ -75,7 +78,7 @@ pub fn fail(url: &str, why: &str) { /// Returns how many went. pub fn trim(dir: &Path, limit: u64) -> usize { let Ok(entries) = std::fs::read_dir(dir) else { return 0 }; - let mut kept: Vec<(std::time::SystemTime, u64, PathBuf)> = entries + let mut kept: Vec<(SystemTime, u64, PathBuf)> = entries .flatten() .map(|e| e.path()) .filter(|p| p.extension().is_none()) diff --git a/crates/ipx/src/auth.rs b/crates/ipx/src/auth.rs index 4fef636..ef765a4 100644 --- a/crates/ipx/src/auth.rs +++ b/crates/ipx/src/auth.rs @@ -6,6 +6,9 @@ use anyhow::{Result, bail}; use argon2::Argon2; use argon2::password_hash::phc::PasswordHash; use argon2::password_hash::{PasswordHasher, PasswordVerifier}; +use std::fs::File; +use std::sync::{Arc, LazyLock}; +use tokio::sync::Semaphore; /// Argon2id with the crate's defaults, which are the OWASP-recommended parameters. The /// salt is generated per password by the hasher itself. @@ -32,17 +35,15 @@ pub fn verify_password(password: &str, stored: &str) -> bool { /// How many password checks run at once: each is tens of milliseconds of CPU, and forty wrong /// passwords at a time, run on the async workers, made every other request wait 4s (#137). /// Half the cores, so a flood of sign-ins waits on itself and the rest of the server has the rest. -static CHECKS: std::sync::LazyLock> = std::sync::LazyLock::new(|| { - std::sync::Arc::new(tokio::sync::Semaphore::new( - std::thread::available_parallelism().map_or(1, |n| (n.get() / 2).max(1)), - )) +static CHECKS: LazyLock> = LazyLock::new(|| { + Arc::new(Semaphore::new(std::thread::available_parallelism().map_or(1, |n| (n.get() / 2).max(1)))) }); /// What a name that is not an account is checked against, so that refusing it takes as long as /// refusing a wrong password does. Refused without a check, it came back 31ms sooner, and the /// time told anyone which names are accounts here (#138). -static DECOY: std::sync::LazyLock = - std::sync::LazyLock::new(|| hash_password("no account here has this password").expect("hashing a fixed password")); +static DECOY: LazyLock = + LazyLock::new(|| hash_password("no account here has this password").expect("hashing a fixed password")); /// A sign-in's password against the account's hash, or against `DECOY` when there is no account /// or it has no password: false either way, in the same time. Off the async workers, and a few at @@ -91,7 +92,7 @@ pub fn api_token_hash(token: &str) -> String { fn getrandom(buf: &mut [u8]) -> std::io::Result<()> { use std::io::Read; - std::fs::File::open("/dev/urandom")?.read_exact(buf) + File::open("/dev/urandom")?.read_exact(buf) } /// A username taken from a proxy header. Cloudflare sends an email address; the local part diff --git a/crates/ipx/src/config.rs b/crates/ipx/src/config.rs index f8d8d8e..f5101a6 100644 --- a/crates/ipx/src/config.rs +++ b/crates/ipx/src/config.rs @@ -3,7 +3,9 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::fs::Permissions; use std::path::{Path, PathBuf}; +use url::Url; #[derive(Debug, Default, Clone, Deserialize, Serialize)] pub struct Config { @@ -319,7 +321,7 @@ impl Feed { } fn same_host(a: &str, b: &str) -> bool { - let host = |u: &str| url::Url::parse(u).ok().and_then(|u| u.host_str().map(str::to_ascii_lowercase)); + let host = |u: &str| Url::parse(u).ok().and_then(|u| u.host_str().map(str::to_ascii_lowercase)); host(a).is_some_and(|h| host(b).as_ref() == Some(&h)) } @@ -456,7 +458,7 @@ fn write_private(path: &Path, text: &str) -> Result<()> { #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?; + std::fs::set_permissions(path, Permissions::from_mode(0o600))?; } Ok(()) } diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 5e6c231..dfec196 100644 --- a/crates/ipx/src/db.rs +++ b/crates/ipx/src/db.rs @@ -7,15 +7,19 @@ use crate::entity::{ }; use anyhow::{Context, Result}; use sea_orm::sea_query::{Expr, Func}; +use sea_orm::sqlx::sqlite::SqliteSynchronous; use sea_orm::{ ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set, Statement, }; +use sea_orm::{ConnectOptions, Database, DatabaseConnection}; use std::path::Path; +use std::time::SystemTime; +use tracing::Instrument; /// A pool of connections, where there was one connection behind a global lock. pub struct Db { - orm: sea_orm::DatabaseConnection, + orm: DatabaseConnection, /// A test's database file, removed when the test is done with it. #[cfg(test)] tmp: Option, @@ -46,7 +50,7 @@ impl Drop for Db { /// here, as the old migrate() did for SQLite, or sea-orm-migration once there are several. A /// database from before 0.7 took its last columns from that migrate(), so it upgrades through /// a 0.7 release first. -async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> { +async fn create_missing(orm: &DatabaseConnection) -> Result<()> { use crate::entity::*; use sea_orm::{ConnectionTrait, Schema}; let schema = Schema::new(orm.get_database_backend()); @@ -241,15 +245,15 @@ fn is_postgres(location: &str) -> bool { /// sqlx's defaults for SQLite are what ipx wants, foreign keys on and a five-second wait for a /// lock, which is what rusqlite was set to, but for one. -async fn connect(location: &str, keep_open: bool) -> Result { - let mut opts = sea_orm::ConnectOptions::new(url_for(location)); +async fn connect(location: &str, keep_open: bool) -> Result { + let mut opts = ConnectOptions::new(url_for(location)); opts.sqlx_logging(false); // NORMAL, SQLite's own advice for WAL: synced at checkpoints, not after every commit. FULL, // the default, waited for the disk after nearly every statement, and a feed's items are // written a row at a time: 1.3s a feed on Tower's /tmp, where one synchronous write takes // 52ms, and 39s to subscribe an admin to 1,500 feeds (#135). A crash of ipx loses nothing; // only a power cut can lose the last transactions, and it never corrupts the file. - opts.map_sqlx_sqlite_opts(|o| o.synchronous(sea_orm::sqlx::sqlite::SqliteSynchronous::Normal)); + opts.map_sqlx_sqlite_opts(|o| o.synchronous(SqliteSynchronous::Normal)); // The same on Postgres, for ipx's own sessions: a commit waited for the WAL to reach the // disk, 12.8ms on the databases project's server, and at about fifty commits a feed a scan // took most of a second a feed (#140). A crash of the Postgres server can lose the last @@ -271,7 +275,7 @@ async fn connect(location: &str, keep_open: bool) -> Result| async move { - tracing::Instrument::instrument(self.rows(sql, args), step(of)) + Instrument::instrument(self.rows(sql, args), step(of)) .await? .iter() .map(|r| Ok((r.try_get::("", "feed_id")?, r.try_get::("", "n")?))) @@ -568,7 +572,7 @@ impl Db { ), vec![user_id.into(), one()], ); - let blocked = tracing::Instrument::instrument( + let blocked = Instrument::instrument( blocklists::Entity::find().filter(blocklists::Column::UserId.eq(user_id)).all(&self.orm), step("blocked"), ); @@ -576,7 +580,7 @@ impl Db { if let Some(id) = only { rows = rows.filter(feeds::Column::Id.eq(id)); } - let rows = tracing::Instrument::instrument(rows.all(&self.orm), step("feeds")); + let rows = Instrument::instrument(rows.all(&self.orm), step("feeds")); let (entries, downloaded, unread, blocked, rows) = tokio::try_join!( entries, downloaded, @@ -819,7 +823,7 @@ impl Db { /// Returns true when this enclosure URL is new. False means we have downloaded it /// before, or deliberately reaped it -- either way it is not fetched again. - pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { + pub async fn mark_downloaded(&self, url: &str, path: &Path, bytes: u64) -> Result<()> { self.exec( "UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3, downloaded_at = $4, last_error = NULL WHERE url = $1", @@ -1002,7 +1006,7 @@ impl Db { .await? .into_iter() .filter_map(|e| Some((e.id, e.path?))) - .filter(|(_, p)| !std::path::Path::new(p).exists()) + .filter(|(_, p)| !Path::new(p).exists()) .collect()) } @@ -3086,7 +3090,7 @@ impl Db { } pub fn now() -> i64 { - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0) + SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0) } #[cfg(test)] diff --git a/crates/ipx/src/deadair.rs b/crates/ipx/src/deadair.rs index 3ea2f56..6ee60fd 100644 --- a/crates/ipx/src/deadair.rs +++ b/crates/ipx/src/deadair.rs @@ -26,6 +26,12 @@ use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; use crate::Ctx; +use axum::body::Bytes; +use std::collections::{HashSet, VecDeque}; +use std::io::SeekFrom; +use std::process::Stdio; +use std::time::{Duration, Instant}; +use tokio::fs::File; /// Quieter than this for longer than `MIN_SILENCE` is a silence: what the measurements on the /// issue used. A podcast's room tone sits well below it and speech well above. @@ -44,7 +50,7 @@ pub const DEFAULT_PAUSE: i64 = 300; const MAX_PIECE: f64 = 10.0; /// How long the maker waits when there is nothing to do: a download or a feed turned on is /// picked up this long after, at most. -const IDLE: std::time::Duration = std::time::Duration::from_secs(120); +const IDLE: Duration = Duration::from_secs(120); /// A span of the original, in milliseconds, that DeadAir leaves out. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -81,28 +87,28 @@ pub fn id3_len(head: &[u8]) -> u64 { pub async fn kept( path: &Path, plan: &Plan, -) -> std::io::Result<(u64, impl futures_util::Stream> + use<>)> { +) -> std::io::Result<(u64, impl futures_util::Stream> + use<>)> { use tokio::io::{AsyncReadExt, AsyncSeekExt}; - let mut file = tokio::fs::File::open(path).await?; + let mut file = File::open(path).await?; let mut head = [0u8; 10]; let n = file.read(&mut head).await?; // Only up to the first frame kept: past the tag can be the Xing frame, which gives the // original's length. let first = plan.pieces.first().map_or(0, |p| p.0); let tag = id3_len(&head[..n]).min(first); - let ranges: std::collections::VecDeque<(u64, u64)> = + let ranges: VecDeque<(u64, u64)> = std::iter::once((0, tag)).chain(plan.pieces.iter().map(|&(o, n, _)| (o, n))).filter(|r| r.1 > 0).collect(); let len = ranges.iter().map(|r| r.1).sum(); let body = futures_util::stream::try_unfold((file, ranges), |(mut f, mut left)| async move { let Some((off, n)) = left.pop_front() else { return Ok(None) }; let mut buf = vec![0; n.min(1 << 16) as usize]; - f.seek(std::io::SeekFrom::Start(off)).await?; + f.seek(SeekFrom::Start(off)).await?; f.read_exact(&mut buf).await?; let got = buf.len() as u64; if got < n { left.push_front((off + got, n - got)); } - Ok(Some((axum::body::Bytes::from(buf), (f, left)))) + Ok(Some((Bytes::from(buf), (f, left)))) }); Ok((len, body)) } @@ -371,7 +377,7 @@ pub async fn run(ctx: Arc, mut stop: tokio::sync::watch::Receiver) { } // Files ffmpeg failed on, passed over until the next start. Not stored: a failure recorded // as the file having nothing to cut was never tried again, even once a fix was deployed (#174). - let mut failed = std::collections::HashSet::new(); + let mut failed = HashSet::new(); loop { // Dropping the pass on a signal kills the ffmpeg it is running (`kill_on_drop`). let done = tokio::select! { @@ -394,7 +400,7 @@ pub async fn run(ctx: Arc, mut stop: tokio::sync::watch::Receiver) { async fn have_ffmpeg() -> bool { tokio::process::Command::new("ffmpeg") .arg("-version") - .stdout(std::process::Stdio::null()) + .stdout(Stdio::null()) .status() .await .is_ok_and(|s| s.success()) @@ -404,7 +410,7 @@ async fn have_ffmpeg() -> bool { /// shortened has already heard through gets none: finished episodes are rarely played again. // shortcut: those stay on the list and are looked at again each pass, one query each; record // them as passed over if a large library makes the pass slow. -async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet) -> Result { +async fn next(ctx: &Ctx, failed: &mut HashSet) -> Result { // 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; @@ -433,7 +439,7 @@ async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet) -> Result< continue; } let src = crate::download::playable(Path::new(&t.path)); - let started = std::time::Instant::now(); + let started = Instant::now(); // 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); @@ -594,7 +600,7 @@ async fn run_out(prog: &str, args: &[&str]) -> Result { fn command(prog: &str) -> tokio::process::Command { let mut c = tokio::process::Command::new(prog); - c.kill_on_drop(true).stdin(std::process::Stdio::null()); + c.kill_on_drop(true).stdin(Stdio::null()); c } diff --git a/crates/ipx/src/directory.rs b/crates/ipx/src/directory.rs index c2c6d49..3377bfc 100644 --- a/crates/ipx/src/directory.rs +++ b/crates/ipx/src/directory.rs @@ -8,9 +8,13 @@ use ipx_directory::{FeedPage, Listing, Me, SnapshotEntry, SnapshotResult}; use serde::de::DeserializeOwned; use crate::{Ctx, Sub, config, feed}; +use std::fmt::{Display, Formatter}; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; +use tracing::Instrument; /// How often the snapshot goes when nothing has woken the sender sooner. -const SHARE_EVERY: std::time::Duration = std::time::Duration::from_secs(30 * 60); +const SHARE_EVERY: Duration = Duration::from_secs(30 * 60); /// What went wrong asking the directory, in words for an admin. #[derive(Debug)] @@ -23,8 +27,8 @@ pub enum DirError { Unreachable(String), } -impl std::fmt::Display for DirError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl Display for DirError { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match self { DirError::NotConfigured => { write!(f, "No directory is set up on this server. An admin sets one in Admin, Directory.") @@ -69,11 +73,8 @@ async fn call( body: Option<&[SnapshotEntry]>, ) -> Result { let (base, token) = target(&ctx.cfg()).ok_or(DirError::NotConfigured)?; - let mut req = ctx - .local_client - .request(method, format!("{base}{path}")) - .bearer_auth(token) - .timeout(std::time::Duration::from_secs(20)); + let mut req = + ctx.local_client.request(method, format!("{base}{path}")).bearer_auth(token).timeout(Duration::from_secs(20)); // The directory's spans join this request's trace (#213). if let Some(tp) = ipx_directory::otel::traceparent() { req = req.header("traceparent", tp); @@ -139,7 +140,7 @@ pub async fn page(ctx: &Ctx, id: &str) -> Result { /// /// shortcut: in memory, one address a listed feed, forgotten on a restart until the Directory is /// next read, which the page does when it loads; a table if that ever shows as initials. -static LISTED_ART: std::sync::LazyLock>> = std::sync::LazyLock::new(Default::default); +static LISTED_ART: LazyLock>> = LazyLock::new(Default::default); fn note_art<'a>(listings: impl Iterator) { let mut art = LISTED_ART.lock().unwrap_or_else(|e| e.into_inner()); @@ -240,10 +241,10 @@ pub async fn share(ctx: &Ctx, entries: &[SnapshotEntry]) -> Result, mut stop: tokio::sync::watch::Receiver) { +pub async fn run(ctx: Arc, mut stop: tokio::sync::watch::Receiver) { // What went last, where, and when: a scan wakes this every half minute or so, and the same // 1,500 feeds went each time, the directory rewriting its rows for nothing (#211). - let mut last: Option<(Vec, std::time::Instant)> = None; + let mut last: Option<(Vec, Instant)> = None; loop { let cfg = ctx.cfg(); if cfg.general.directory_share && configured(&cfg) { @@ -254,9 +255,9 @@ pub async fn run(ctx: std::sync::Arc, mut stop: tokio::sync::watch::Receive if forced || resend(last.as_ref(), &key) { let span = tracing::info_span!(target: "ipx::directory", "directory_share", feeds = entries.len()); - let got = tracing::Instrument::instrument(share(&ctx, &entries), span).await; + let got = Instrument::instrument(share(&ctx, &entries), span).await; if got.is_ok() { - last = Some((key, std::time::Instant::now())); + last = Some((key, Instant::now())); } Some(got) } else { @@ -294,7 +295,7 @@ pub async fn run(ctx: std::sync::Arc, mut stop: tokio::sync::watch::Receive /// Whether to send the snapshot `key` stands for: one that changed, or went to another directory /// or with another token, goes now; the same again only once SHARE_EVERY has passed, so a /// directory that lost it, or dropped this server's counts, has it back within the half hour. -fn resend(last: Option<&(Vec, std::time::Instant)>, key: &[u8]) -> bool { +fn resend(last: Option<&(Vec, Instant)>, key: &[u8]) -> bool { last.is_none_or(|(k, at)| k != key || at.elapsed() >= SHARE_EVERY) } diff --git a/crates/ipx/src/download.rs b/crates/ipx/src/download.rs index 5d180c5..ac24f0c 100644 --- a/crates/ipx/src/download.rs +++ b/crates/ipx/src/download.rs @@ -6,6 +6,9 @@ use std::path::{Path, PathBuf}; use tokio::io::AsyncWriteExt; use crate::config::{Config, Feed as FeedCfg, Organize}; +use chrono::Local; +use tokio::fs::File; +use url::Url; /// Forbidden characters that were separating words: they become "-" so the words stay /// apart. The original's stringCleaning() deleted them, turning "Show | Series" into @@ -83,7 +86,7 @@ pub fn filename_for(url: &str, content_disposition: Option<&str>) -> String { if let Some(name) = content_disposition.and_then(disposition_filename) { return sanitize(&name); } - let from_url = url::Url::parse(url) + let from_url = Url::parse(url) .ok() .and_then(|u| u.path_segments().and_then(|mut s| s.rfind(|p| !p.is_empty())).map(str::to_owned)) .map(|seg| percent_encoding::percent_decode_str(&seg).decode_utf8_lossy().into_owned()) @@ -151,7 +154,7 @@ pub async fn download( tokio::fs::create_dir_all(&tmp_dir).await.with_context(|| format!("creating {}", tmp_dir.display()))?; let tmp = tmp_dir.join(&name); - let mut file = tokio::fs::File::create(&tmp).await.with_context(|| format!("creating {}", tmp.display()))?; + let mut file = File::create(&tmp).await.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); @@ -246,7 +249,7 @@ async fn sniff(path: &Path) -> Result { async fn read_head(path: &Path, n: usize) -> Result> { use tokio::io::AsyncReadExt; - let mut f = tokio::fs::File::open(path).await?; + let mut f = File::open(path).await?; let mut buf = vec![0u8; n]; let read = f.read(&mut buf).await?; buf.truncate(read); @@ -277,7 +280,7 @@ fn unique_path(dir: &Path, name: &str) -> PathBuf { /// letting the sanitizer eat the separator. pub fn folder_for(cfg: &Config, id: &str, feed_cfg: &FeedCfg, title: Option<&str>) -> String { match cfg.general.organize { - Organize::Date => chrono::Local::now().format("%m-%d-%Y").to_string(), + Organize::Date => Local::now().format("%m-%d-%Y").to_string(), Organize::Feed => { let raw = feed_cfg.folder.as_deref().or(title).filter(|s| !s.trim().is_empty()).unwrap_or(id); raw.split('/') diff --git a/crates/ipx/src/ipc.rs b/crates/ipx/src/ipc.rs index 327c30d..cd85fcf 100644 --- a/crates/ipx/src/ipc.rs +++ b/crates/ipx/src/ipc.rs @@ -2,7 +2,12 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use std::future::Future; use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::{broadcast, mpsc}; @@ -124,7 +129,7 @@ impl Event { /// Whether a page signed in as this person is sent it. Everyone had every event, so a key in /// a paid feed's path, which `redacted` keeps, reached people who do not read the feed (#325). /// An admin has them all, for the log. - pub fn is_for(&self, user: &crate::db::User, mine: &std::collections::HashSet) -> bool { + pub fn is_for(&self, user: &crate::db::User, mine: &HashSet) -> bool { if user.is_admin { return true; } @@ -350,8 +355,7 @@ pub async fn daemon_is_live(path: &Path) -> bool { /// the meantime. The answer goes to the client that asked and no one else: broadcast, it ended /// any `ipx fetch` that was watching a scan, since `status` is a terminal event. /// A future, since reading the counts is a database query. -pub type StatusFn = - std::sync::Arc std::pin::Pin + Send>> + Send + Sync>; +pub type StatusFn = Arc Pin + Send>> + Send + Sync>; /// Accepts connections, feeding commands to `cmds` and events from `events` back out. pub async fn serve( @@ -455,9 +459,7 @@ pub async fn reload(path: &Path) -> Result<()> { } anyhow::bail!("the daemon closed the connection without answering") }; - tokio::time::timeout(std::time::Duration::from_secs(10), answer) - .await - .context("the daemon did not answer in 10s")? + tokio::time::timeout(Duration::from_secs(10), answer).await.context("the daemon did not answer in 10s")? } /// Sends one command to a running daemon and prints the events it produces. diff --git a/crates/ipx/src/logbuf.rs b/crates/ipx/src/logbuf.rs index 75eac6c..ee0a3dc 100644 --- a/crates/ipx/src/logbuf.rs +++ b/crates/ipx/src/logbuf.rs @@ -4,7 +4,9 @@ //! to read. Capturing inside the tracing pipeline works the same either way. use std::collections::VecDeque; +use std::fmt::Debug; use std::sync::{LazyLock, Mutex}; +use tracing::Subscriber; use tracing::field::{Field, Visit}; use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; @@ -67,7 +69,7 @@ pub fn since(after: u64, limit: usize) -> (Vec, u64) { /// A tracing layer that mirrors every event into the ring. pub struct RingLayer; -impl Layer for RingLayer { +impl Layer for RingLayer { fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) { let mut v = Collect::default(); event.record(&mut v); @@ -102,7 +104,7 @@ impl Collect { } impl Visit for Collect { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + fn record_debug(&mut self, field: &Field, value: &dyn Debug) { self.add(field, format!("{value:?}")); } // Numbers and bools reach record_debug through the trait's defaults, which prints them the diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 1ec54b2..5faebc1 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -18,12 +18,29 @@ mod torrent; mod web; use anyhow::{Context, Result}; +use chrono::DateTime; use clap::{Parser, Subcommand}; use ipc::{Command as Cmd, Emitter, Event}; +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::fs::OpenOptions; use std::future::Future; +use std::io::IsTerminal; +use std::ops::{Deref, DerefMut}; +use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicI64}; +use std::time::{Duration, Instant}; +use tokio::runtime::Handle; +use tokio::sync::{MutexGuard, Notify, OnceCell, Semaphore}; use tokio::sync::{broadcast, mpsc, watch}; +use tokio::task::JoinHandle; +use tracing::Subscriber; +use tracing_subscriber::EnvFilter; +use tracing_subscriber::fmt::format::JsonFields; +use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields}; +use tracing_subscriber::registry::LookupSpan; +use url::Url; #[derive(Parser)] #[command(name = "ipx", version, about = "A headless podcatcher")] @@ -122,7 +139,7 @@ enum UserCmd { /// `admin-password` beside config.toml, readable by the daemon's user alone. Every install began /// as admin / ipodderx, published in the README, and nothing made anyone change it (#321). /// Returns the password and where it can be found, for the log, which never shows it. -fn first_password(config_dir: &std::path::Path, from_env: Option) -> Result<(String, String)> { +fn first_password(config_dir: &Path, from_env: Option) -> Result<(String, String)> { if let Some(p) = from_env.filter(|p| !p.is_empty()) { return Ok((p, "with the password in IPX_ADMIN_PASSWORD".into())); } @@ -133,12 +150,12 @@ fn first_password(config_dir: &std::path::Path, from_env: Option) -> Res } /// Writes a secret only its owner may read: created 0600, never readable for a moment between. -fn write_private(file: &std::path::Path, text: &str) -> Result<()> { +fn write_private(file: &Path, text: &str) -> Result<()> { use std::io::Write; use std::os::unix::fs::OpenOptionsExt; - std::fs::create_dir_all(file.parent().unwrap_or(std::path::Path::new(".")))?; + std::fs::create_dir_all(file.parent().unwrap_or(Path::new(".")))?; let _ = std::fs::remove_file(file); - let mut f = std::fs::OpenOptions::new() + let mut f = OpenOptions::new() .write(true) .create_new(true) .mode(0o600) @@ -153,7 +170,7 @@ pub struct Ctx { /// Swapped wholesale when the web UI rewrites config.toml, so a running daemon picks /// up feed changes without a restart. Callers take a snapshot; no guard is ever held /// across an await. - pub cfg: std::sync::RwLock>, + pub cfg: std::sync::RwLock>, pub db: db::Db, /// For anything a feed or a person names: refuses internal addresses (#320). pub client: reqwest::Client, @@ -166,9 +183,9 @@ pub struct Ctx { pub out: Emitter, /// Started on first use: a BitTorrent session binds ports and starts a DHT, which is /// rude to do for a config that has never seen a torrent. - pub torrents: tokio::sync::OnceCell, + pub torrents: OnceCell, /// Caps how many torrents run at once when they are detached. - pub torrent_slots: Arc, + pub torrent_slots: Arc, /// Needed because a subscribed OPML rewrites the feed list as it syncs. pub config_path: PathBuf, /// Run torrents off the command worker. A torrent takes minutes to fetch metadata and @@ -179,7 +196,7 @@ pub struct Ctx { /// Feeds whose host could not be looked up for the moment, and when to try them again: a /// failure that says nothing about the feed, kept off it and out of the database (#143). /// Forgotten on a restart, which only makes the retry sooner. - pub retry: std::sync::Mutex>, + pub retry: std::sync::Mutex>, /// Held from taking a copy of the config to storing it (`edit_cfg`). Each change stores the /// whole catalogue from its copy, and two at once stored over each other: a feed added while /// another add spent a minute fetching was gone once that one stored (#153). @@ -189,37 +206,37 @@ pub struct Ctx { pub cfg_write: tokio::sync::Mutex<()>, /// Wakes DeadAir's maker when a show is turned on, so its copies start at once instead of /// at the next pass (#171). - pub deadair: tokio::sync::Notify, + pub deadair: Notify, /// Wakes the Signals judge when there are new items or a Signal is made or changed (#254). - pub signals: tokio::sync::Notify, + pub signals: Notify, /// People whose TypeSafe key was refused, passed over until they save another. - pub signals_refused: std::sync::Mutex>, + pub signals_refused: std::sync::Mutex>, /// When the Directory was last cleared of dead feeds (`clean_directory`), 0 for not since /// starting. - pub directory_cleaned: std::sync::atomic::AtomicI64, + pub directory_cleaned: AtomicI64, /// Sends the snapshot to the directory now, as when its settings change (#205). - pub directory_wake: tokio::sync::Notify, + pub directory_wake: Notify, /// When the snapshot last went and what came of it, for the admin page. pub directory_shared: std::sync::Mutex>, /// Send the snapshot even if it has not changed: after an unblock or a dropped snapshot the /// directory needs it again, though nothing here differs (#206). - pub directory_resend: std::sync::atomic::AtomicBool, + pub directory_resend: AtomicBool, } /// A copy of the config to change, holding off every other change until `store_cfg` keeps it. pub struct CfgEdit<'a> { cfg: config::Config, - _held: tokio::sync::MutexGuard<'a, ()>, + _held: MutexGuard<'a, ()>, } -impl std::ops::Deref for CfgEdit<'_> { +impl Deref for CfgEdit<'_> { type Target = config::Config; fn deref(&self) -> &config::Config { &self.cfg } } -impl std::ops::DerefMut for CfgEdit<'_> { +impl DerefMut for CfgEdit<'_> { fn deref_mut(&mut self) -> &mut config::Config { &mut self.cfg } @@ -261,7 +278,7 @@ impl Ctx { self.db.set_feed_error(id, url, msg).await } - pub fn cfg(&self) -> std::sync::Arc { + pub fn cfg(&self) -> Arc { self.cfg.read().expect("a panic elsewhere poisoned this lock").clone() } @@ -282,7 +299,7 @@ impl Ctx { } fn set_cfg(&self, cfg: config::Config) { - *self.cfg.write().expect("a panic elsewhere poisoned this lock") = std::sync::Arc::new(cfg); + *self.cfg.write().expect("a panic elsewhere poisoned this lock") = Arc::new(cfg); tracing::info!("config reloaded"); // A feed added, removed, made private or given a category, or the directory's own // settings changed: the directory hears of it now, not at the next half hour (#205). @@ -312,25 +329,23 @@ async fn main() -> Result<()> { // Two filters, deliberately different. stderr follows IPX_LOG; the in-app buffer // keeps debug as well, so the log view can show protocol traffic and routine // skips that would be noise on a terminal. IPX_UI_LOG overrides it. - let stderr_filter = - || tracing_subscriber::EnvFilter::try_from_env("IPX_LOG").unwrap_or_else(|_| "ipx=info".into()); + let stderr_filter = || EnvFilter::try_from_env("IPX_LOG").unwrap_or_else(|_| "ipx=info".into()); // One JSON object a line for Loki (#91), with each event's fields as its own; text // otherwise, for someone reading a terminal. let json = std::env::var("IPX_LOG_FORMAT").is_ok_and(|f| f.eq_ignore_ascii_case("json")); - let ui_filter = - tracing_subscriber::EnvFilter::try_from_env("IPX_UI_LOG").unwrap_or_else(|_| "ipx=debug".into()); + let ui_filter = EnvFilter::try_from_env("IPX_UI_LOG").unwrap_or_else(|_| "ipx=debug".into()); tracing_subscriber::registry() .with((!json).then(|| { tracing_subscriber::fmt::layer() .with_writer(std::io::stderr) // Colour for a terminal only: in docker logs and Loki the escapes are noise // every query has to strip (#88). - .with_ansi(std::io::IsTerminal::is_terminal(&std::io::stderr())) + .with_ansi(IsTerminal::is_terminal(&std::io::stderr())) .with_filter(stderr_filter()) })) .with(json.then(|| { tracing_subscriber::fmt::layer() - .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new()) + .fmt_fields(JsonFields::new()) .event_format(WithTrace( tracing_subscriber::fmt::format() .json() @@ -346,7 +361,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,ipx_feed=info")) + .with_filter(EnvFilter::new("ipx=info,ipx_feed=info")) })) .init(); } @@ -390,24 +405,24 @@ async fn main() -> Result<()> { let (events, _) = broadcast::channel(1024); let guard = feed::guard::Guard::new(feed::guard::Allowed::parse(&cfg.general.fetch_internal)?); let ctx = Arc::new(Ctx { - cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)), + cfg: std::sync::RwLock::new(Arc::new(cfg)), db, client: reqwest::Client::builder() .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) // Connecting only, so it bounds a download's start, not a long download (#108). - .connect_timeout(std::time::Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)) // Two minutes without a byte is a stalled download, however long it is: unbounded, one // held the worker, and every scan behind it, for good, while the daemon read healthy (#152). - .read_timeout(std::time::Duration::from_secs(120)) + .read_timeout(Duration::from_secs(120)) .connector_layer(guard.clone()) .build()?, local_client: reqwest::Client::builder() .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) - .connect_timeout(std::time::Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)) .build()?, feed_client: reqwest::Client::builder() .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) - .connect_timeout(std::time::Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)) .read_timeout(feed::FEED_SILENCE) .redirect(reqwest::redirect::Policy::none()) .connector_layer(guard) @@ -417,8 +432,8 @@ async fn main() -> Result<()> { } else { Emitter::terminal() }, - torrents: tokio::sync::OnceCell::new(), - torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)), + torrents: OnceCell::new(), + torrent_slots: Arc::new(Semaphore::new(2)), config_path: config_path.clone(), detach_torrents: is_daemon, retry: Default::default(), @@ -463,15 +478,15 @@ async fn main() -> Result<()> { /// the object it writes. A line outside any traced span, or with no trace exporter, is unchanged. struct WithTrace(F); -impl tracing_subscriber::fmt::FormatEvent for WithTrace +impl FormatEvent for WithTrace where - F: tracing_subscriber::fmt::FormatEvent, - S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, - N: for<'w> tracing_subscriber::fmt::FormatFields<'w> + 'static, + F: FormatEvent, + S: Subscriber + for<'a> LookupSpan<'a>, + N: for<'w> FormatFields<'w> + 'static, { fn format_event( &self, - ctx: &tracing_subscriber::fmt::FmtContext<'_, S, N>, + ctx: &FmtContext<'_, S, N>, mut w: tracing_subscriber::fmt::format::Writer<'_>, ev: &tracing::Event<'_>, ) -> std::fmt::Result { @@ -533,7 +548,7 @@ async fn user_cmd(ctx: &Arc, cmd: UserCmd) -> Result<()> { for u in users { let added = u .created - .and_then(|t| chrono::DateTime::from_timestamp(t, 0)) + .and_then(|t| DateTime::from_timestamp(t, 0)) .map_or("?".into(), |d| d.format("%Y-%m-%d").to_string()); let seen = u.last_login.map_or("never signed in".into(), |t| format!("signed in {}", ago(Some(t)))); println!( @@ -652,7 +667,7 @@ async fn daemon( // A database with nobody in it cannot be signed into. if ctx.db.users().await?.is_empty() { - let dir = ctx.config_path.parent().unwrap_or(std::path::Path::new(".")); + let dir = ctx.config_path.parent().unwrap_or(Path::new(".")); let (password, told) = first_password(dir, std::env::var("IPX_ADMIN_PASSWORD").ok())?; ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?; tracing::warn!( @@ -732,7 +747,7 @@ async fn daemon( { let (ctx, mut stop) = (ctx.clone(), rx_stop.clone()); tokio::spawn(async move { - let mut every = tokio::time::interval(std::time::Duration::from_secs(30)); + let mut every = tokio::time::interval(Duration::from_secs(30)); loop { tokio::select! { _ = every.tick() => { status(&ctx).await; } @@ -766,7 +781,7 @@ async fn daemon( if *stop.borrow() { break; } - let wait = if std::mem::take(&mut first) { std::time::Duration::ZERO } else { until_next_scan(&ctx).await }; + let wait = if std::mem::take(&mut first) { Duration::ZERO } else { until_next_scan(&ctx).await }; tokio::select! { biased; _ = stop.changed() => break, @@ -805,11 +820,11 @@ async fn daemon( /// Starts the web UI when it is switched on, minting and saving a token if there is none. async fn start_web( ctx: &Arc, - config_path: &std::path::Path, + config_path: &Path, web_addr: Option, cmds: &mpsc::Sender, events: &broadcast::Sender, -) -> Result>> { +) -> Result>> { let cfg = ctx.cfg(); let enabled = cfg.web.enabled || web_addr.is_some(); if !enabled { @@ -866,8 +881,8 @@ async fn start_web( cmds: cmds.clone(), events: events.clone(), playing: playing.clone(), - runtime: tokio::runtime::Handle::current(), - started: std::time::Instant::now(), + runtime: Handle::current(), + started: Instant::now(), }; let at = metrics::spawn(&cfg.web.metrics_bind, sources)?; tracing::info!(%at, "serving metrics for Prometheus at /metrics"); @@ -991,7 +1006,7 @@ async fn add_one( // Slugs must be unique across derived feeds too, or a new feed can collide with one // an OPML already introduced. - let mut taken: std::collections::BTreeMap = + let mut taken: BTreeMap = subscriptions(ctx).await?.into_iter().map(|s| (s.id, s.cfg)).collect(); // A removed feed keeps its rows, so its id is only free again for the same feed: re-adding // it gets its history back, and a different feed does not inherit someone else's. @@ -1007,7 +1022,7 @@ async fn add_one( /// Host plus last path segment, for naming a feed we could not read. fn url_stem(url: &str) -> String { - url::Url::parse(url).ok().and_then(|u| u.host_str().map(str::to_owned)).unwrap_or_else(|| url.to_owned()) + Url::parse(url).ok().and_then(|u| u.host_str().map(str::to_owned)).unwrap_or_else(|| url.to_owned()) } async fn rm(ctx: &Ctx, feed: &str) -> Result<()> { @@ -1026,7 +1041,7 @@ async fn rm(ctx: &Ctx, feed: &str) -> Result<()> { Ok(()) } -async fn import(ctx: &Ctx, file: &std::path::Path) -> Result<()> { +async fn import(ctx: &Ctx, file: &Path) -> Result<()> { let text = std::fs::read_to_string(file).with_context(|| format!("reading {}", file.display()))?; // The CLI speaks for the operator, as the shared web token does. let admin = ctx @@ -1070,7 +1085,7 @@ pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i6 let mut cfg = ctx.edit_cfg().await; // By address, once: each feed looked through every subscription and the whole catalogue, and // the catalogue's lock is held throughout (#338). - let mut known = std::collections::HashMap::new(); + let mut known = HashMap::new(); for s in subscriptions(ctx).await? { // The first with an address wins, as a search in order found it. known.entry(s.cfg.url).or_insert(s.id); @@ -1131,7 +1146,7 @@ pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i6 /// database for the feeds and the server settings (issue #18). The first time a database holds /// neither, it takes them from config.toml, which is then cut down to the rest, the original kept /// beside it as config.toml.pre-database. -async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &std::path::Path) -> Result { +async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &Path) -> Result { for _ in 0..2 { if let Some((stored, feeds)) = db.stored_config().await? { if config::Config::file_holds_stored(path) { @@ -2142,9 +2157,8 @@ async fn sync_group(ctx: &Arc, parent_id: &str, listed: &[(String, String)] let mut added = vec![]; // Read once, not once a new feed: each read is the whole feeds table. let known = ctx.db.feed_urls().await?; - let fixed: std::collections::HashSet<&str> = - cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect(); - let known_ids: std::collections::HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect(); + let fixed: HashSet<&str> = cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect(); + let known_ids: HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect(); for (title, url) in listed { // Already known, whether derived or promoted into the config. @@ -2325,7 +2339,7 @@ async fn fetch_one( enclosure: i64, feed_cfg: &config::Feed, url: &str, - dest_dir: &std::path::Path, + dest_dir: &Path, ) -> Result> { // Throttled to whole percents, as the original's lastDLStepSize guard did. let mut last_pct = -1i64; @@ -2485,7 +2499,7 @@ async fn torrent_one( feed_id: &str, enclosure: i64, url: &str, - dest_dir: &std::path::Path, + dest_dir: &Path, ) -> Result<(PathBuf, u64)> { let name = download::filename_for(url, None); let mut last_pct = -1i64; diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index c75ee8a..6be96fe 100644 --- a/crates/ipx/src/metrics.rs +++ b/crates/ipx/src/metrics.rs @@ -6,10 +6,16 @@ //! Prometheus exporters have none: checking a token is a database lookup. Written out by hand: //! the text format is a few lines, and a client library would be a dependency for them. +use axum::Router; +use std::collections::BTreeMap; use std::fmt::Write as _; +use std::net::SocketAddr; use std::sync::Arc; use std::sync::atomic::Ordering::Relaxed; use std::sync::atomic::{AtomicI64, AtomicU64}; +use std::time::Instant; +use tokio::runtime::Handle; +use tower_http::compression::CompressionLayer; /// 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 @@ -51,11 +57,11 @@ pub struct Stats { /// What each person's Signals, icons and key checks used, and whose key paid (#278): items /// 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>, + people: std::sync::Mutex>, /// The worker's job now, and since when. - job: std::sync::Mutex>, + job: std::sync::Mutex>, /// Jobs finished, by kind: how many, and their seconds all told. - jobs: std::sync::Mutex>, + jobs: std::sync::Mutex>, } impl Stats { @@ -80,9 +86,9 @@ impl Stats { jev_calls: [const { AtomicU64::new(0) }; 5], jev_micros: AtomicU64::new(0), jev_paced_micros: AtomicU64::new(0), - people: std::sync::Mutex::new(std::collections::BTreeMap::new()), + people: std::sync::Mutex::new(BTreeMap::new()), job: std::sync::Mutex::new(None), - jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()), + jobs: std::sync::Mutex::new(BTreeMap::new()), } } @@ -99,7 +105,7 @@ impl Stats { /// 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().expect("a panic elsewhere poisoned this lock") = Some((kind, std::time::Instant::now())); + *self.job.lock().expect("a panic elsewhere poisoned this lock") = Some((kind, Instant::now())); Job(self) } } @@ -143,12 +149,12 @@ pub struct Sources { pub events: tokio::sync::broadcast::Sender, pub playing: crate::deadair::Playing, /// The daemon's runtime, not the metrics thread's own. - pub runtime: tokio::runtime::Handle, - pub started: std::time::Instant, + pub runtime: Handle, + pub started: Instant, } /// Starts serving /metrics on `bind`, on a thread of its own, and says where. -pub fn spawn(bind: &str, s: Sources) -> anyhow::Result { +pub fn spawn(bind: &str, s: Sources) -> anyhow::Result { use anyhow::Context as _; // Bound here, so a port already taken is the daemon's error at start, not a thread's later. let listener = std::net::TcpListener::bind(bind).with_context(|| format!("binding metrics to {bind}"))?; @@ -161,7 +167,7 @@ pub fn spawn(bind: &str, s: Sources) -> anyhow::Result { Err(e) => return tracing::error!(error = %e, "metrics could not start"), }; rt.block_on(async move { - let app = axum::Router::new() + let app = Router::new() .route( "/metrics", axum::routing::get(move || { @@ -170,7 +176,7 @@ pub fn spawn(bind: &str, s: Sources) -> anyhow::Result { }), ) // Prometheus asks for gzip, every 15 s (#253). - .layer(tower_http::compression::CompressionLayer::new().gzip(true)); + .layer(CompressionLayer::new().gzip(true)); match tokio::net::TcpListener::from_std(listener) { Ok(l) => { if let Err(e) = axum::serve(l, app).await { diff --git a/crates/ipx/src/retention.rs b/crates/ipx/src/retention.rs index 5bcaa22..3fbce94 100644 --- a/crates/ipx/src/retention.rs +++ b/crates/ipx/src/retention.rs @@ -6,6 +6,8 @@ use tracing::Instrument; use crate::config::Config; use crate::db; use crate::db::{Candidate, Db}; +use std::io::ErrorKind; +use std::path::Path; /// Headroom kept below the quota, as getToQuota()'s `1048576 * 50` did. pub const PAD_BYTES: u64 = 50 * 1024 * 1024; @@ -103,8 +105,8 @@ async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result { // A torrent's folder measures as a directory entry, not what is in it: the recorded size then. let size = std::fs::metadata(&c.path).ok().filter(|m| m.is_file()).map_or(c.bytes.max(0) as u64, |m| m.len()); // Missing is fine: the point is that it is gone. - if let Err(e) = crate::download::remove(std::path::Path::new(&c.path)) - && e.kind() != std::io::ErrorKind::NotFound + if let Err(e) = crate::download::remove(Path::new(&c.path)) + && e.kind() != ErrorKind::NotFound { tracing::warn!(path = c.path, error = %e, "could not delete"); return Ok(0); diff --git a/crates/ipx/src/signals.rs b/crates/ipx/src/signals.rs index 9aae1ba..d2e2edd 100644 --- a/crates/ipx/src/signals.rs +++ b/crates/ipx/src/signals.rs @@ -19,12 +19,16 @@ use crate::db; use crate::db::{Ask, Unjudged}; use crate::feed; use crate::metrics::STATS; +use std::collections::BTreeMap; +use std::fmt::{Display, Formatter}; +use std::future::Future; +use std::time::Duration; /// Items put to Jev a pass. A pass ends early once there are none left, so this only says how /// long the queue is held before people who got a key later are looked at. const BATCH: i64 = 50; /// Between passes when nothing wakes it: a scan or a new Signal does, so this is a backstop. -const IDLE: std::time::Duration = std::time::Duration::from_secs(300); +const IDLE: Duration = Duration::from_secs(300); /// Calls to Jev in flight at once (#275). One at a time, about 150 ms each, a new Signal's month /// of items took twenty minutes. Enough for `MAX_RATE` at that pace; the pace is `pace`'s to keep. const IN_FLIGHT: usize = 10; @@ -103,7 +107,7 @@ async fn pass(ctx: &Ctx) -> Result { tracing::Span::current().record("calls", groups.len()); let refused = std::sync::Mutex::new(HashSet::new()); let stopped = AtomicBool::new(false); - let mut hits_for = std::collections::BTreeMap::::new(); + let mut hits_for = BTreeMap::::new(); let mut judged = 0; // Each call starts only when a place is free, so one that sees a key refused or the pass // stopped is not sent; those already in flight are let finish, and their answers kept. @@ -203,8 +207,8 @@ struct Refused { body: String, } -impl std::fmt::Display for Refused { - fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { +impl Display for Refused { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { write!(f, "TypeSafe answered {}: {}", self.status, self.body) } } @@ -238,7 +242,7 @@ fn slot( tokens: f64, ) -> (tokio::time::Instant, tokio::time::Instant) { let start = next.max(now); - (start, start + std::time::Duration::from_secs_f64((1.0 / MAX_RATE).max(tokens / MAX_TOKENS))) + (start, start + Duration::from_secs_f64((1.0 / MAX_RATE).max(tokens / MAX_TOKENS))) } /// Waits for this call's turn. Its tokens are guessed from the request, a token to three bytes @@ -261,7 +265,7 @@ async fn pace(body: &serde_json::Value) { /// TypeSafe said too many: nobody calls again until it says, or a second. async fn back_off(retry_after: Option) { - let until = tokio::time::Instant::now() + std::time::Duration::from_secs(retry_after.unwrap_or(1).min(60)); + let until = tokio::time::Instant::now() + Duration::from_secs(retry_after.unwrap_or(1).min(60)); let mut next = NEXT_CALL.lock().await; *next = Some(next.map_or(until, |n| n.max(until))); } @@ -435,7 +439,7 @@ fn usage(answer: &serde_json::Value) -> [f64; 2] { /// 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. -async fn called(span: tracing::Span, call: impl std::future::Future>) -> Result { +async fn called(span: tracing::Span, call: impl Future>) -> Result { let started = std::time::Instant::now(); let r = call.instrument(span.clone()).await; STATS.jev_micros.fetch_add(started.elapsed().as_micros() as u64, Relaxed); diff --git a/crates/ipx/src/torrent.rs b/crates/ipx/src/torrent.rs index aaf65f8..f44ad40 100644 --- a/crates/ipx/src/torrent.rs +++ b/crates/ipx/src/torrent.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use crate::config::Config; +use librqbit::{ListenerOptions, ManagedTorrent}; pub struct Torrents { session: Arc, @@ -26,7 +27,7 @@ impl Torrents { let listen_addr: SocketAddr = format!("0.0.0.0:{lo}").parse()?; let opts = SessionOptions { - listen: Some(librqbit::ListenerOptions { listen_addr, ..Default::default() }), + listen: Some(ListenerOptions { listen_addr, ..Default::default() }), client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()), ..Default::default() }; @@ -127,7 +128,7 @@ impl Torrents { } /// Seeds until the ratio or the time limit is reached, whichever comes first. - async fn seed(&self, cfg: &Config, handle: &Arc) { + async fn seed(&self, cfg: &Config, handle: &Arc) { if cfg.torrent.seed_ratio <= 0.0 || cfg.torrent.seed_time_mins == 0 { return; } diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index 03bbc34..e2599b4 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -23,7 +23,16 @@ use tower_http::services::ServeFile; use crate::Ctx; use crate::db; use crate::ipc::{Command, Event}; +use axum::Extension; +use axum::extract::{ConnectInfo, FromRequestParts}; +use axum::http::request::Parts; +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::net::SocketAddr; +use std::sync::LazyLock; +use std::time::{Duration, Instant}; +use tokio::net::TcpListener; use tracing::Instrument; +use url::Url; const COOKIE: &str = "ipx_token"; @@ -98,14 +107,13 @@ pub async fn serve(state: WebState, bind: &str) -> Result<()> { use axum::serve::ListenerExt; // Nagle off, as axum 0.7 had it: under load a response goes out in parts, and with Nagle // on its last small part waits for the browser's delayed ACK, up to 40 ms (#170). - let listener = - tokio::net::TcpListener::bind(bind).await.with_context(|| format!("binding {bind}"))?.tap_io(|tcp| { - if let Err(e) = tcp.set_nodelay(true) { - tracing::debug!(error = %e, "could not set TCP_NODELAY"); - } - }); + let listener = TcpListener::bind(bind).await.with_context(|| format!("binding {bind}"))?.tap_io(|tcp| { + if let Err(e) = tcp.set_nodelay(true) { + tracing::debug!(error = %e, "could not set TCP_NODELAY"); + } + }); tracing::info!(bind, "web ui listening"); - axum::serve(listener, router(state).into_make_service_with_connect_info::()) + axum::serve(listener, router(state).into_make_service_with_connect_info::()) .await .context("serving the web ui") } @@ -240,7 +248,7 @@ async fn vouched_name( state: &WebState, cfg: &crate::config::Config, peer: String, - headers: &axum::http::HeaderMap, + headers: &HeaderMap, ) -> Option { if cfg.web.trusted_header.is_empty() || !cfg.web.trusted_proxies.iter().any(|p| p == &peer) { return None; @@ -257,21 +265,15 @@ async fn vouched_name( } fn peer(req: &Request) -> String { - req.extensions() - .get::>() - .map(|c| c.0.ip().to_string()) - .unwrap_or_default() + req.extensions().get::>().map(|c| c.0.ip().to_string()).unwrap_or_default() } /// Handlers take `User` to say they need one; the auth layer put it there, and nothing /// reaches a handler without passing through it. -impl axum::extract::FromRequestParts for crate::db::User { +impl FromRequestParts for crate::db::User { type Rejection = (StatusCode, &'static str); - async fn from_request_parts( - parts: &mut axum::http::request::Parts, - _state: &S, - ) -> std::result::Result { + async fn from_request_parts(parts: &mut Parts, _state: &S) -> std::result::Result { parts.extensions.get::().cloned().ok_or((StatusCode::UNAUTHORIZED, "sign in")) } } @@ -346,7 +348,7 @@ async fn logout(State(state): State, req: Request) -> Response { async fn me( State(state): State, user: crate::db::User, - axum::Extension(Proxied(by_proxy)): axum::Extension, + Extension(Proxied(by_proxy)): Extension, ) -> Json { let url = state.ctx.cfg().web.sign_out_url.clone(); let sign_out = (by_proxy && !url.is_empty()).then_some(url); @@ -502,17 +504,12 @@ struct IconAsk { /// the route is not tied to a Signal, so a loop on it ran up the server's bill and held back /// everyone's Signals behind the shared pace (#340). A Signal is named a few times at most. const ICON_PICKS_AN_HOUR: u32 = 20; -static ICON_PICKS: std::sync::LazyLock>> = - std::sync::LazyLock::new(Default::default); +static ICON_PICKS: LazyLock>> = LazyLock::new(Default::default); /// Counts a pick against the person's hour, which starts at their first pick in it. -fn icon_pick_allowed( - picks: &mut std::collections::HashMap, - user: i64, - now: std::time::Instant, -) -> bool { +fn icon_pick_allowed(picks: &mut HashMap, user: i64, now: Instant) -> bool { let (start, used) = picks.entry(user).or_insert((now, 0)); - if now.duration_since(*start) >= std::time::Duration::from_secs(3600) { + if now.duration_since(*start) >= Duration::from_secs(3600) { (*start, *used) = (now, 0); } if *used >= ICON_PICKS_AN_HOUR { @@ -535,7 +532,7 @@ async fn pick_signal_icon( if !icon_pick_allowed( &mut ICON_PICKS.lock().expect("a panic elsewhere poisoned this lock"), user.id, - std::time::Instant::now(), + Instant::now(), ) { return Err(ApiError::too_many("enough icons picked for one hour; choose one by hand")); } @@ -899,7 +896,7 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>) // Config entries plus the feeds derived from OPML subscriptions -- the catalogue. // What comes back is only the part of it this person subscribes to. let subs = crate::subscriptions(&state.ctx).instrument(db::step("catalogue")).await?; - let mine: std::collections::HashMap = state + let mine: HashMap = state .ctx .db .subscriptions_for(user.id) @@ -913,9 +910,8 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>) let kept = state.ctx.db.kept_feeds(user.id).instrument(db::step("kept")).await?; let (hidden, quiet) = state.ctx.db.hidden_from(user.id).instrument(db::step("hidden_from")).await?; let prefs = state.ctx.db.deadair_prefs(Some(user.id)).await?; - let deadair: std::collections::HashMap<&str, bool> = prefs.iter().map(|(_, f, w, _)| (f.as_str(), *w)).collect(); - let pauses: std::collections::HashMap<&str, i64> = - prefs.iter().filter_map(|(_, f, _, p)| Some((f.as_str(), (*p)?))).collect(); + let deadair: HashMap<&str, bool> = prefs.iter().map(|(_, f, w, _)| (f.as_str(), *w)).collect(); + let pauses: HashMap<&str, i64> = prefs.iter().filter_map(|(_, f, _, p)| Some((f.as_str(), (*p)?))).collect(); let mut listed = state.ctx.db.feed_list(user.id, only).instrument(db::step("feed_list")).await?; let mut out = Vec::with_capacity(mine.len()); let asked = subs.iter().filter(|s| only.is_none_or(|o| o == s.id)); @@ -1034,7 +1030,7 @@ fn dir_error(e: crate::directory::DirError) -> ApiError { /// address, since the directory and this server name a feed differently. async fn rows_for(state: &WebState, user_id: i64, listed: Vec) -> Result> { // A step each, as the feed list has: a Directory page of 983 ms showed 19 under spans (#250). - let mine: std::collections::HashSet = state + let mine: HashSet = state .ctx .db .subscriptions_for(user_id) @@ -1043,7 +1039,7 @@ async fn rows_for(state: &WebState, user_id: i64, listed: Vec = crate::subscriptions(&state.ctx) + let here: HashMap = crate::subscriptions(&state.ctx) .instrument(db::step("catalogue")) .await? .into_iter() @@ -1232,16 +1228,12 @@ async fn subscribe_popular( /// Validates a replacement feed URL: present, parseable, and not already subscribed under /// a different id. Returns the trimmed URL. -fn check_url( - url: &str, - id: &str, - feeds: &std::collections::BTreeMap, -) -> Result { +fn check_url(url: &str, id: &str, feeds: &BTreeMap) -> Result { let url = url.trim(); if url.is_empty() { return Err("the feed URL cannot be empty".into()); } - match url::Url::parse(url) { + match Url::parse(url) { Ok(u) if u.scheme() == "http" || u.scheme() == "https" => {} Ok(u) => return Err(format!("{:?} is not an http(s) URL", u.scheme())), Err(e) => return Err(format!("that is not a valid URL: {e}")),