diff --git a/Cargo.toml b/Cargo.toml index 9e1c50e..f6b60d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,10 @@ resolver = "3" [workspace.package] edition = "2024" +# None of ours needs unsafe, and the style guide asks for none where safe code will do. +[workspace.lints.rust] +unsafe_code = "forbid" + [workspace.dependencies] ipx-feed = { path = "crates/feed" } ipx-directory = { path = "crates/directory" } diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index e88e6ce..4cf16e2 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -8,6 +8,9 @@ publish = false name = "ipx-directory" path = "src/main.rs" +[lints] +workspace = true + [dependencies] ammonia.workspace = true anyhow.workspace = true diff --git a/crates/directory/src/config.rs b/crates/directory/src/config.rs index dcb8fbf..89124f4 100644 --- a/crates/directory/src/config.rs +++ b/crates/directory/src/config.rs @@ -123,18 +123,21 @@ impl Config { } /// A token's SHA-256 in hex, as the config file holds it. +#[must_use] pub fn hash(token: &str) -> String { use sha2::Digest; sha2::Sha256::digest(token.trim().as_bytes()).iter().map(|b| format!("{b:02x}")).collect() } /// Equal without saying, by how long it took, how much of the guess was right. +#[must_use] pub fn same_secret(a: &str, b: &str) -> bool { let (a, b) = (a.as_bytes(), b.as_bytes()); a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 } /// A new token: `ipxd_` and 256 random bits, the prefix so one found in a log says what it is. +#[must_use] pub fn new_token() -> String { use std::io::Read; let mut bytes = [0u8; 32]; diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index a3f054d..cc5b532 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -180,14 +180,14 @@ impl Directory { /// Takes the tokens of a config file read again, and drops the snapshots of any that went. pub async fn set_tokens(&self, tokens: Vec) -> anyhow::Result { let names: Vec = tokens.iter().map(|t| t.name.clone()).collect(); - *self.tokens.write().unwrap() = tokens; + *self.tokens.write().expect("a panic elsewhere poisoned this lock") = tokens; self.store.keep_instances(&names).await } 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| { + self.tokens.read().expect("a panic elsewhere poisoned this lock").iter().fold(None, |found, t| { if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found } }) } diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 2880f2c..5233dfd 100644 --- a/crates/directory/src/main.rs +++ b/crates/directory/src/main.rs @@ -2,7 +2,8 @@ use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; -use ipx_directory::{Directory, config, store::Store}; +use ipx_directory::store::Store; +use ipx_directory::{Directory, config}; use tracing_subscriber::prelude::*; #[derive(Parser)] @@ -101,7 +102,7 @@ async fn serve(path: &std::path::Path) -> Result<()> { .ok() .filter(|u| !u.trim().is_empty()) .with_context(|| format!("{} is not set: the directory needs a database URL", cfg.database_url_env))?; - let store = Store::open(&url, true).await.context("opening the database")?; + let store = Store::open(&url, /* keep_open */ true).await.context("opening the database")?; let dir = Directory::new(store, vec![], ipx_feed::guard::Allowed::parse(&cfg.fetch_internal)?)?; let dropped = dir.set_tokens(tokens).await?; diff --git a/crates/directory/src/metrics.rs b/crates/directory/src/metrics.rs index 90a0cf0..44d8999 100644 --- a/crates/directory/src/metrics.rs +++ b/crates/directory/src/metrics.rs @@ -12,6 +12,7 @@ use std::sync::atomic::Ordering::Relaxed; static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; /// The process's resident memory in bytes and its open files, from /proc; None elsewhere. +#[must_use] pub fn process() -> (Option, Option) { let rss = std::fs::read_to_string("/proc/self/status").ok().and_then(|s| { let kb = diff --git a/crates/directory/src/otel.rs b/crates/directory/src/otel.rs index bdf5fd1..e8a722f 100644 --- a/crates/directory/src/otel.rs +++ b/crates/directory/src/otel.rs @@ -23,6 +23,7 @@ pub fn provider(service: &'static str) -> anyhow::Result) -> Option { use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId, TraceState}; let v = match headers.get("traceparent") { @@ -46,6 +47,7 @@ pub fn remote_parent(headers: &axum::http::HeaderMap, query: Option<&str>) -> Op /// The current span as a `traceparent`, for a request to another service, so its spans join /// this trace. None outside a traced span. +#[must_use] pub fn traceparent() -> Option { use opentelemetry::trace::TraceContextExt; use tracing_opentelemetry::OpenTelemetrySpanExt; diff --git a/crates/directory/src/store.rs b/crates/directory/src/store.rs index c46e160..4579f13 100644 --- a/crates/directory/src/store.rs +++ b/crates/directory/src/store.rs @@ -35,12 +35,14 @@ pub enum Refreshed { Failed(String), } +#[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) } /// 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()?; if !matches!(u.scheme(), "http" | "https") || u.host_str().is_none() { @@ -52,6 +54,7 @@ pub fn normalise(url: &str) -> Option { /// A feed's id: the first 64 bits of its normalised address's SHA-256, so every instance, and /// anyone else, arrives at the same id for the same feed without asking. +#[must_use] pub fn feed_id(normalised: &str) -> String { use sha2::Digest; sha2::Sha256::digest(normalised.as_bytes())[..8].iter().map(|b| format!("{b:02x}")).collect() @@ -274,7 +277,8 @@ impl Store { .map(|r| r.try_get("", "instance")) .collect::>()?; // Apple's charts are no token's, and stay until the next sync replaces them. - for name in present.iter().filter(|n| !names.contains(n) && n.as_str() != crate::apple::INSTANCE) { + let gone = present.iter().filter(|n| !names.contains(n) && n.as_str() != crate::apple::INSTANCE); + for name in gone { dropped += self.drop_instance(name).await?; } Ok(dropped) diff --git a/crates/feed/Cargo.toml b/crates/feed/Cargo.toml index 8114acb..7115df9 100644 --- a/crates/feed/Cargo.toml +++ b/crates/feed/Cargo.toml @@ -4,6 +4,9 @@ version = "0.0.0" publish = false edition.workspace = true +[lints] +workspace = true + [dependencies] anyhow.workspace = true atom_syndication.workspace = true diff --git a/crates/feed/src/guard.rs b/crates/feed/src/guard.rs index 3fe53c3..1b273d9 100644 --- a/crates/feed/src/guard.rs +++ b/crates/feed/src/guard.rs @@ -28,6 +28,7 @@ impl Allowed { list.iter().map(|s| parse_net(s)).collect::>>().map(|v| Self(Arc::new(v))) } + #[must_use] pub fn permits(&self, ip: IpAddr) -> bool { let ip = canonical(ip); !is_internal(ip) || self.0.iter().any(|&(net, bits)| contains(net, bits, ip)) @@ -73,6 +74,7 @@ fn contains(net: IpAddr, bits: u8, ip: IpAddr) -> bool { /// unstable. // shortcut: 6to4 (2002::/16) and Teredo addresses that wrap an internal IPv4 one pass; check the // embedded address if an IPv6 route to them ever exists here. +#[must_use] pub fn is_internal(ip: IpAddr) -> bool { match canonical(ip) { IpAddr::V4(a) => { @@ -107,6 +109,7 @@ pub fn is_internal(ip: IpAddr) -> bool { pub struct Guard(Allowed); impl Guard { + #[must_use] pub fn new(allowed: Allowed) -> Self { Self(allowed) } diff --git a/crates/feed/src/lib.rs b/crates/feed/src/lib.rs index f1fc771..0275011 100644 --- a/crates/feed/src/lib.rs +++ b/crates/feed/src/lib.rs @@ -87,6 +87,7 @@ pub static FETCHED_BYTES: std::sync::atomic::AtomicU64 = std::sync::atomic::Atom /// An address as a trace may keep it: without its query or user and password, where a private /// feed carries its credential (Patreon's `?auth=`), which Tempo kept for anyone reading it (#225). /// shortcut: a token in the path (`/feed/.xml`) stays; strip paths too if a feed does that. +#[must_use] pub fn redact(url: &str) -> String { match url::Url::parse(url) { Ok(mut u) => { @@ -103,6 +104,7 @@ pub fn redact(url: &str) -> String { /// Text with each address in it redacted, for what is logged and sent to every open page: an /// error from reqwest names the whole address it fetched, a private feed's key with it (#238). +#[must_use] pub fn redact_text(text: &str) -> String { let mut out = String::with_capacity(text.len()); let mut rest = text; @@ -263,6 +265,7 @@ pub struct Failure { /// A failure's kind, for the log's `error.type` (#91): the HTTP status where there is one, as /// OpenTelemetry names an HTTP error, and otherwise a word for what went wrong. Matches the /// same wording as `explain_failure`; a message it does not know is "other", never a wrong kind. +#[must_use] pub fn failure_kind(msg: &str) -> (String, Option) { let low = msg.to_ascii_lowercase(); let code = low.split("http ").skip(1).find_map(|r| r.get(..3)?.parse::().ok()); @@ -344,6 +347,7 @@ pub fn explain_failure(msg: &str) -> Option { /// "Temporary failure in name resolution", musl's "Try again" -- as against a name that does /// not exist ("Name or service not known"). It says the network was not answering, not that the /// feed is gone (#143). +#[must_use] pub fn is_temporary_lookup_failure(msg: &str) -> bool { let low = msg.to_ascii_lowercase(); low.contains("dns error") && (low.contains("temporary failure in name resolution") || low.contains("try again")) @@ -353,6 +357,7 @@ pub fn is_temporary_lookup_failure(msg: &str) -> bool { /// /// The original matched on the URL ending in ".opml" (iPXClass.py:34), which misses an /// OPML served from a URL without that extension. Sniffing the body catches both. +#[must_use] pub fn is_opml(bytes: &[u8]) -> bool { let head = &bytes[..bytes.len().min(1024)]; let text = String::from_utf8_lossy(head).to_lowercase(); @@ -407,6 +412,7 @@ pub fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, Strin } /// The of an OPML document. +#[must_use] pub fn opml_title(bytes: &[u8]) -> Option<String> { let text = String::from_utf8_lossy(bytes); let doc = read_opml(&text).ok()?; @@ -427,6 +433,7 @@ fn patreon_parts(url: &str) -> Option<(String, Option<String>)> { } /// A Patreon link naming a creator but no show. +#[must_use] pub fn is_patreon_creator(url: &str) -> bool { matches!(patreon_parts(url), Some((_, None))) } @@ -434,6 +441,7 @@ pub fn is_patreon_creator(url: &str) -> bool { /// What was typed into Add feed, as a URL. A bare Patreon token is taken as its creator's /// feed, since the token alone says whose it is. An address with no scheme is https: 'cnn.com' /// was stored as typed and every check failed with "relative URL without a base" (#101). +#[must_use] pub fn expand_input(input: &str) -> String { let s = input.trim(); let token = s.len() >= 20 && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'); @@ -451,6 +459,7 @@ pub fn expand_input(input: &str) -> String { /// Whether two URLs are the same feed. One Patreon show has several spellings -- by the /// creator's name, by number, or with no creator at all -- and the token and show are what /// identify it. +#[must_use] pub fn same_feed(a: &str, b: &str) -> bool { // Parsing both is the slow part, and only Patreon's addresses can differ and still match. a == b @@ -1019,6 +1028,7 @@ fn drop_player_repeats(encs: &mut Vec<Enclosure>) { /// An enclosure URL without WordPress's player number, for telling repeats of one file apart /// from different files. +#[must_use] pub fn same_file_key(url: &str) -> String { let Ok(mut u) = url::Url::parse(url) else { return url.to_owned() }; let kept: Vec<(String, String)> = u @@ -1095,6 +1105,7 @@ fn is_markdown(s: &str) -> bool { } /// Markdown made HTML, uncleaned: whoever shows it cleans it as any feed's HTML. +#[must_use] pub fn markdown_html(md: &str) -> String { use pulldown_cmark::{Options, Parser, html}; let mut out = String::new(); @@ -1302,10 +1313,12 @@ const ALIASES: &[(&str, &str)] = /// category and then its subcategories as Apple's does (#118). A podcast's own is stored as its /// subcategory, and one an admin typed that is not Apple's stands as a category of its own. /// Apple's categories as ipx has them, for the directory to check against Apple's own (#252). +#[must_use] pub fn categories() -> &'static [(&'static str, &'static [&'static str])] { CATEGORIES } +#[must_use] pub fn category_path(name: &str) -> (String, Option<String>) { // A feed's own text, which can end in a stray comma: "Soccer," was a category of its own. let name = name.trim().trim_end_matches([',', ';', '.']).trim_end(); @@ -1376,6 +1389,7 @@ fn picked(answer: &serde_json::Value) -> Result<String> { /// /// ponytail: a heuristic. A key hidden in the path of a host not in `PAID_HOSTS` gets /// through; a per-feed private flag is the upgrade (#205). +#[must_use] pub fn url_looks_private(feed_url: &str) -> bool { let Ok(u) = url::Url::parse(feed_url) else { return true }; // Paid-feed services put the subscriber's key in the path as often as in the query, and a diff --git a/crates/ipx/Cargo.toml b/crates/ipx/Cargo.toml index 92206c2..9b0311a 100644 --- a/crates/ipx/Cargo.toml +++ b/crates/ipx/Cargo.toml @@ -4,6 +4,9 @@ version = "1.5.1-dev" edition.workspace = true publish = false +[lints] +workspace = true + [dependencies] ipx-feed.workspace = true ipx-directory.workspace = true diff --git a/crates/ipx/src/access.rs b/crates/ipx/src/access.rs index a612b3b..7934af4 100644 --- a/crates/ipx/src/access.rs +++ b/crates/ipx/src/access.rs @@ -11,7 +11,7 @@ use std::time::{Duration, Instant}; use anyhow::{Context, Result}; use jsonwebtoken::jwk::JwkSet; -use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode, decode_header}; +use jsonwebtoken::{Algorithm, DecodingKey, Validation}; /// Cloudflare rotates its keys every six weeks or so, publishing the new one before using it. /// A token naming a key not seen yet refetches, but no more often than this, so a stream of @@ -55,7 +55,7 @@ impl Keys { F: FnOnce() -> Fut, Fut: Future<Output = Result<JwkSet>>, { - let kid = decode_header(token).ok()?.kid?; + let kid = jsonwebtoken::decode_header(token).ok()?.kid?; let key = match self.key(&kid) { Some(k) => k, None => { @@ -78,7 +78,7 @@ impl Keys { v.set_audience(&[aud]); v.set_issuer(&[format!("https://{team}")]); v.validate_nbf = true; - match decode::<Claims>(token, &key, &v) { + match jsonwebtoken::decode::<Claims>(token, &key, &v) { Ok(data) => crate::auth::name_from_header(&data.claims.email?), Err(e) => { tracing::warn!(error = %e, "refused a Cloudflare Access token"); @@ -88,12 +88,12 @@ impl Keys { } fn key(&self, kid: &str) -> Option<DecodingKey> { - self.cache.lock().unwrap().keys.get(kid).cloned() + self.cache.lock().expect("a panic elsewhere poisoned this lock").keys.get(kid).cloned() } /// Takes the slot as it answers, so two requests at once do not both fetch. fn may_refetch(&self) -> bool { - let mut c = self.cache.lock().unwrap(); + let mut c = self.cache.lock().expect("a panic elsewhere poisoned this lock"); if c.fetched.is_some_and(|t| t.elapsed() < REFETCH_EVERY) { return false; } @@ -105,7 +105,7 @@ impl Keys { fn store(&self, set: JwkSet) { let keys = set.keys.iter().filter_map(|k| Some((k.common.key_id.clone()?, DecodingKey::from_jwk(k).ok()?))).collect(); - let mut c = self.cache.lock().unwrap(); + let mut c = self.cache.lock().expect("a panic elsewhere poisoned this lock"); c.keys = keys; c.fetched = Some(Instant::now()); } @@ -128,7 +128,7 @@ async fn fetch(client: &reqwest::Client, team: &str) -> Result<JwkSet> { #[cfg(test)] mod tests { use super::*; - use jsonwebtoken::{EncodingKey, Header, encode, get_current_timestamp}; + use jsonwebtoken::{EncodingKey, Header}; const TEAM: &str = "team.cloudflareaccess.com"; const AUD: &str = "aud-tag"; @@ -141,11 +141,11 @@ mod tests { let mut h = Header::new(alg); h.kid = Some("k1".into()); let k = if alg == Algorithm::RS256 { EncodingKey::from_rsa_der(key) } else { EncodingKey::from_secret(key) }; - encode(&h, &claims, &k).unwrap() + jsonwebtoken::encode(&h, &claims, &k).unwrap() } fn claims(aud: &str, exp_in: i64) -> serde_json::Value { - let now = get_current_timestamp() as i64; + let now = jsonwebtoken::get_current_timestamp() as i64; serde_json::json!({ "aud": [aud], "iss": format!("https://{TEAM}"), "email": "Rays@SDF1.net", "iat": now, "nbf": now, "exp": now + exp_in, "type": "app", @@ -201,7 +201,7 @@ mod tests { // A made-up key id straight after: not fetched again. let mut h = Header::new(Algorithm::RS256); h.kid = Some("nobody".into()); - let stray = encode(&h, &claims(AUD, 300), &EncodingKey::from_rsa_der(SIGNER)).unwrap(); + let stray = jsonwebtoken::encode(&h, &claims(AUD, 300), &EncodingKey::from_rsa_der(SIGNER)).unwrap(); let count = || { fetches.fetch_add(1, std::sync::atomic::Ordering::SeqCst); async { Ok(jwks()) } diff --git a/crates/ipx/src/auth.rs b/crates/ipx/src/auth.rs index facb2bd..4fef636 100644 --- a/crates/ipx/src/auth.rs +++ b/crates/ipx/src/auth.rs @@ -4,7 +4,8 @@ use anyhow::{Result, bail}; use argon2::Argon2; -use argon2::password_hash::{PasswordHasher, PasswordVerifier, phc::PasswordHash}; +use argon2::password_hash::phc::PasswordHash; +use argon2::password_hash::{PasswordHasher, PasswordVerifier}; /// Argon2id with the crate's defaults, which are the OWASP-recommended parameters. The /// salt is generated per password by the hasher itself. diff --git a/crates/ipx/src/config.rs b/crates/ipx/src/config.rs index 04c4fb8..f8d8d8e 100644 --- a/crates/ipx/src/config.rs +++ b/crates/ipx/src/config.rs @@ -527,7 +527,7 @@ pub fn unique_slug(text: &str, taken: impl Fn(&str) -> bool) -> String { if !taken(&base) { return base; } - (2..).map(|n| format!("{base}-{n}")).find(|s| !taken(s)).unwrap() + (2..).map(|n| format!("{base}-{n}")).find(|s| !taken(s)).expect("an unending count finds a free name") } /// `$var`, or `~/fallback` when it is unset or empty, as the XDG base directory spec says. @@ -756,8 +756,9 @@ mod tests { }; assert_eq!(f.password().as_deref(), Some("literal")); - unsafe { std::env::set_var("IPX_TEST_PASS", "from-env") }; - f.password_env = Some("IPX_TEST_PASS".into()); - assert_eq!(f.password().as_deref(), Some("from-env")); + // A variable every test run already has: setting one is unsafe while other tests' + // threads read the environment. + f.password_env = Some("PATH".into()); + assert_eq!(f.password(), std::env::var("PATH").ok(), "password_env wins over password"); } } diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index fbe2178..5e6c231 100644 --- a/crates/ipx/src/db.rs +++ b/crates/ipx/src/db.rs @@ -475,7 +475,6 @@ impl Db { Ok(Self { orm, tmp: Some(path) }) } - #[cfg(test)] /// The first column of every row, for a test to check what a query left behind. #[cfg(test)] pub async fn i64s_for_test(&self, sql: &str) -> Vec<i64> { diff --git a/crates/ipx/src/deadair.rs b/crates/ipx/src/deadair.rs index b5e71e3..3ea2f56 100644 --- a/crates/ipx/src/deadair.rs +++ b/crates/ipx/src/deadair.rs @@ -20,7 +20,7 @@ use std::sync::Arc; use tracing::Instrument; -use crate::db::step; +use crate::db; use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; @@ -341,7 +341,11 @@ pub async fn on_save( secs: i64, duration: Option<i64>, ) -> Result<(i64, Option<i64>)> { - let plan = playing.lock().unwrap().get(&(user, feed.to_owned(), guid.to_owned())).cloned(); + let plan = playing + .lock() + .expect("a panic elsewhere poisoned this lock") + .get(&(user, feed.to_owned(), guid.to_owned())) + .cloned(); let Some(p) = plan else { return Ok((secs, duration)) }; let original = |s: i64| to_original(&p.cuts, s * 1000) / 1000; let (secs, duration) = (original(secs), duration.map(original)); @@ -446,7 +450,7 @@ async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result< }; ctx.db .store_plans(t.id, &serde_json::to_string(&plans)?) - .instrument(span.in_scope(|| step("store_plans"))) + .instrument(span.in_scope(|| db::step("store_plans"))) .await?; let usual = plans.get(&DEFAULT_PAUSE); span.record("cuts", usual.map_or(0, |p| p.cuts.len())); @@ -474,7 +478,7 @@ async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result< /// anything but MP3 or AAC on its own (ADTS; AAC in MP4 has no byte ranges a playlist can /// use), or one with no silence long enough to cut. pub async fn analyse(src: &Path) -> Result<Option<BTreeMap<i64, Plan>>> { - let Some((first, frame)) = probe(src).instrument(step("probe")).await? else { return Ok(None) }; + let Some((first, frame)) = probe(src).instrument(db::step("probe")).await? else { return Ok(None) }; let path = src.display().to_string(); let log = ffmpeg(&[ "-hide_banner", @@ -489,11 +493,11 @@ pub async fn analyse(src: &Path) -> Result<Option<BTreeMap<i64, Plan>>> { "null", "-", ]) - .instrument(step("silencedetect")) + .instrument(db::step("silencedetect")) .await .context("finding the silences")?; let silences = parse_silences(&log); - let frames = frame_bytes(&path).instrument(step("frame_bytes")).await?; + let frames = frame_bytes(&path).instrument(db::step("frame_bytes")).await?; let plans: BTreeMap<i64, Plan> = PAUSES .iter() .map(|&pause| { diff --git a/crates/ipx/src/directory.rs b/crates/ipx/src/directory.rs index 0088da2..c2c6d49 100644 --- a/crates/ipx/src/directory.rs +++ b/crates/ipx/src/directory.rs @@ -279,7 +279,8 @@ pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receive Ok(r) => (Some(r), None), Err(e) => (None, Some(e.to_string())), }; - *ctx.directory_shared.lock().unwrap() = Some(Shared { at: crate::db::now(), result, error }); + *ctx.directory_shared.lock().expect("a panic elsewhere poisoned this lock") = + Some(Shared { at: crate::db::now(), result, error }); } } tokio::select! { diff --git a/crates/ipx/src/download.rs b/crates/ipx/src/download.rs index 6b009f5..5d180c5 100644 --- a/crates/ipx/src/download.rs +++ b/crates/ipx/src/download.rs @@ -312,7 +312,8 @@ pub fn blocked(words: &[String], haystacks: &[&str]) -> bool { // at either end: " new york " is in " the new york times ", " york " is not in " yorkshire ". let norm = |s: &str| { let mut out = String::from(" "); - for w in s.split(|c: char| !c.is_alphanumeric()).filter(|w| !w.is_empty()) { + let words = s.split(|c: char| !c.is_alphanumeric()).filter(|w| !w.is_empty()); + for w in words { out.push_str(&w.to_lowercase()); out.push(' '); } diff --git a/crates/ipx/src/ipc.rs b/crates/ipx/src/ipc.rs index 5824824..327c30d 100644 --- a/crates/ipx/src/ipc.rs +++ b/crates/ipx/src/ipc.rs @@ -262,21 +262,20 @@ impl Event { /// terminal. Addresses lose their query and password, where a private feed keeps its key, and /// so do the ones in an error's words (#238). fn redacted(mut self) -> Self { - use crate::feed::{redact, redact_text}; match &mut self { Event::DownloadError { url, msg, .. } => { - *url = redact(url); - *msg = redact_text(msg); + *url = crate::feed::redact(url); + *msg = crate::feed::redact_text(msg); } // Progress and DownloadDone too: a Patreon file's address carries its signature (#292). Event::TorrentDeferred { url, .. } | Event::Progress { url, .. } | Event::DownloadDone { url, .. } => { - *url = redact(url) + *url = crate::feed::redact(url) } Event::FeedMoved { from, to, .. } => { - *from = redact(from); - *to = redact(to); + *from = crate::feed::redact(from); + *to = crate::feed::redact(to); } - Event::FeedError { msg, .. } | Event::Error { msg } => *msg = redact_text(msg), + Event::FeedError { msg, .. } | Event::Error { msg } => *msg = crate::feed::redact_text(msg), _ => {} } self @@ -424,7 +423,7 @@ async fn handle( Ok(cmd @ (Command::Status | Command::Reload)) => { tracing::info!(target: "ipx::io", "-> {line}"); let ev = status(cmd).await; - log_event(&ev, true); + log_event(&ev, /* wire */ true); let _ = reply.send(ev).await; } Ok(cmd) => { diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 6b3689a..1ec54b2 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -239,7 +239,7 @@ impl Ctx { subscribed: bool, ) -> Option<i64> { let at = due_at(cfg, feed, state, subscribed); - match self.retry.lock().unwrap().get(id) { + match self.retry.lock().expect("a panic elsewhere poisoned this lock").get(id) { Some(&retry) => Some(at.unwrap_or(0).max(retry)), None => at, } @@ -252,14 +252,17 @@ impl Ctx { /// instead, with nothing on the feed to say it failed. async fn feed_failed(&self, id: &str, url: &str, msg: &str) -> Result<()> { if feed::is_temporary_lookup_failure(msg) { - self.retry.lock().unwrap().insert(id.to_owned(), db::now() + RETRY_LOOKUP_SECS); + self.retry + .lock() + .expect("a panic elsewhere poisoned this lock") + .insert(id.to_owned(), db::now() + RETRY_LOOKUP_SECS); return Ok(()); } self.db.set_feed_error(id, url, msg).await } pub fn cfg(&self) -> std::sync::Arc<config::Config> { - self.cfg.read().unwrap().clone() + self.cfg.read().expect("a panic elsewhere poisoned this lock").clone() } /// The config as it stands, to change and then keep with `store_cfg`; any other change waits @@ -279,7 +282,7 @@ impl Ctx { } fn set_cfg(&self, cfg: config::Config) { - *self.cfg.write().unwrap() = std::sync::Arc::new(cfg); + *self.cfg.write().expect("a panic elsewhere poisoned this lock") = std::sync::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). @@ -350,7 +353,9 @@ async fn main() -> Result<()> { let config_path = cli.config.clone().unwrap_or_else(config::config_path); let cfg = config::Config::load(&config_path)?; - let db = db::Db::open(&db::location(), matches!(cli.command, Command::Daemon { .. })).await?; + // The daemon holds its connections open; a one-off command opens what it uses. + let keep_open = matches!(cli.command, Command::Daemon { .. }); + let db = db::Db::open(&db::location(), keep_open).await?; // A daemon owns the state; don't have two processes downloading the same thing. let wire_cmd = match &cli.command { @@ -407,7 +412,11 @@ async fn main() -> Result<()> { .redirect(reqwest::redirect::Policy::none()) .connector_layer(guard) .build()?, - out: if is_daemon { Emitter::socket(events.clone(), false) } else { Emitter::terminal() }, + out: if is_daemon { + Emitter::socket(events.clone(), /* print */ false) + } else { + Emitter::terminal() + }, torrents: tokio::sync::OnceCell::new(), torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)), config_path: config_path.clone(), @@ -584,12 +593,12 @@ async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> { match cmd { Cmd::Fetch { feed, force, feeds } => { // Make room before pulling more down, as the original did per download. - reap(ctx, false, false).await?; + reap(ctx, /* dry_run */ false, /* standalone */ false).await?; fetch(ctx, feed.as_deref(), force, &feeds).await?; metrics::STATS.last_scan_done.store(db::now(), std::sync::atomic::Ordering::Relaxed); Ok(()) } - Cmd::Reap { dry_run } => reap(ctx, dry_run, true).await, + Cmd::Reap { dry_run } => reap(ctx, dry_run, /* standalone */ true).await, Cmd::Download { enclosure } => download_one(ctx, enclosure).await, Cmd::Status => { ctx.out.emit(status(ctx).await); @@ -645,7 +654,7 @@ async fn daemon( if ctx.db.users().await?.is_empty() { let dir = ctx.config_path.parent().unwrap_or(std::path::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)?), true).await?; + ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?; tracing::warn!( "no accounts yet: created 'admin' {told}. Change it with `echo -n <password> | ipx user passwd admin`" ); @@ -872,8 +881,9 @@ async fn start_web( } async fn shutdown() { - use tokio::signal::unix::{SignalKind, signal}; - let mut term = match signal(SignalKind::terminate()) { + use tokio::signal::unix; + use tokio::signal::unix::SignalKind; + let mut term = match unix::signal(SignalKind::terminate()) { Ok(s) => s, Err(_) => return std::future::pending().await, }; @@ -1253,7 +1263,8 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { tracing::debug!(gone, "trimmed the artwork kept on disk"); } } - for c in r.aged_out.iter().chain(r.over_quota.iter()) { + let removed = r.aged_out.iter().chain(r.over_quota.iter()); + for c in removed { ctx.out.emit(Event::Reaped { path: c.path.clone(), bytes: c.bytes.max(0) as u64 }); } if standalone { @@ -1280,7 +1291,8 @@ async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool, scope: &[String] |s: &Sub| scope.is_empty() || scope.contains(&s.id) || s.cfg.group.as_ref().is_some_and(|g| scope.contains(g)); let mut states = ctx.db.http_states().await?; let mut due = vec![]; - for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id) && in_scope(s)) { + let asked = subs.iter().filter(|s| only.is_none_or(|o| o == s.id) && in_scope(s)); + for sub in asked { let id = &sub.id; let mut state = states.remove(id).unwrap_or_default(); // A scan someone asked for reads the feed in full. With the validators it only skipped the @@ -1491,12 +1503,9 @@ pub async fn leave(ctx: &Ctx, user_id: i64, id: &str) -> Result<()> { ctx.db.unsubscribe(user_id, id).instrument(db::step("unsubscribe")).await?; // What they kept stays on their list, a feed of its own once nobody reads the list (#280). let kept = ctx.db.kept_feeds(user_id).instrument(db::step("kept")).await?; - for child in subscriptions(ctx) - .instrument(db::step("subscriptions")) - .await? - .iter() - .filter(|s| s.cfg.group.as_deref() == Some(id) && !kept.contains(&s.id)) - { + let all = subscriptions(ctx).instrument(db::step("subscriptions")).await?; + let unkept = all.iter().filter(|s| s.cfg.group.as_deref() == Some(id) && !kept.contains(&s.id)); + for child in unkept { ctx.db.unsubscribe(user_id, &child.id).instrument(db::step("unsubscribe")).await?; } if ctx.db.subscriber_counts().instrument(db::step("subscriber_counts")).await?.contains_key(id) { @@ -1545,7 +1554,9 @@ pub async fn remove_user(ctx: &Ctx, user_id: i64) -> Result<()> { /// it was removed, None when that is no longer known. pub async fn retire_group(ctx: &Ctx, parent_id: &str, parent: Option<&config::Feed>) -> Result<()> { let cfg = ctx.cfg(); - for m in ctx.db.managed_feeds().await?.into_iter().filter(|m| m.group_id == parent_id) { + let managed = ctx.db.managed_feeds().await?; + let derived = managed.into_iter().filter(|m| m.group_id == parent_id); + for m in derived { if cfg.feeds.contains_key(&m.id) { // Scanned from its config entry and still read. Dropped as derived, its stored // entries would go with it: davewiner's 11 were promoted without being unmanaged. @@ -1624,6 +1635,7 @@ async fn retire_stranded(ctx: &Ctx) -> Result<usize> { /// When a feed is next due, or None for one never checked, which is due now. A feed nobody /// subscribes to, one listed in the Directory, is read once a day: enough to keep its entry /// current, without fetching it hourly for no one. +#[must_use] pub fn due_at(cfg: &config::Config, feed: &config::Feed, state: &db::HttpState, subscribed: bool) -> Option<i64> { let last = state.last_checked?; let floor = if subscribed { 0 } else { 86_400 }; @@ -2156,15 +2168,13 @@ async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(String, String)] // Whoever subscribes to the OPML subscribes to what it lists: that is what taking a // subscription means. Their own feeds are untouched. - for id in ctx - .db - .managed_feeds() - .await? + let managed = ctx.db.managed_feeds().await?; + let list_and_feeds = managed .iter() .filter(|m| m.group_id == parent_id) .map(|m| m.id.clone()) - .chain(std::iter::once(parent_id.to_string())) - { + .chain(std::iter::once(parent_id.to_string())); + for id in list_and_feeds { for user in ctx.db.users().await? { if ctx.db.subscription(user.id, parent_id).await?.is_some() { ctx.db.subscribe(user.id, &id).await?; @@ -2176,7 +2186,8 @@ async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(String, String)] let mut removed = 0; let mut kept = 0; let parent = ctx.cfg().feeds.get(parent_id).cloned(); - for m in existing.iter().filter(|m| m.group_id == parent_id) { + let in_group = existing.iter().filter(|m| m.group_id == parent_id); + for m in in_group { if listed.iter().any(|(_, u)| u == &m.url) { continue; } @@ -2768,7 +2779,8 @@ mod tests { let at = metrics::spawn("127.0.0.1:0", sources).unwrap(); let text = ctx.client.get(format!("http://{at}/metrics")).send().await.unwrap().text().await.unwrap(); // Every sample a name, optional labels and a number, as Prometheus parses them (#194). - for line in text.lines().filter(|l| !l.starts_with('#')) { + let samples = text.lines().filter(|l| !l.starts_with('#')); + for line in samples { let (name, value) = line.rsplit_once(' ').unwrap(); assert!(name.starts_with("ipx_") && value.parse::<f64>().is_ok(), "{line}"); } diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index 52e8d46..c75ee8a 100644 --- a/crates/ipx/src/metrics.rs +++ b/crates/ipx/src/metrics.rs @@ -8,7 +8,8 @@ use std::fmt::Write as _; use std::sync::Arc; -use std::sync::atomic::{AtomicI64, AtomicU64, Ordering::Relaxed}; +use std::sync::atomic::Ordering::Relaxed; +use std::sync::atomic::{AtomicI64, AtomicU64}; /// 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 @@ -88,7 +89,7 @@ impl Stats { /// Adds to what `who` used on `key` ("server" or "own"): items judged, fits, input and output /// tokens. pub fn bill(&self, who: &str, key: &'static str, used: [f64; 4]) { - let mut people = self.people.lock().unwrap(); + let mut people = self.people.lock().expect("a panic elsewhere poisoned this lock"); let row = people.entry((who.to_owned(), key)).or_default(); for (r, u) in row.iter_mut().zip(used) { *r += u; @@ -98,7 +99,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().unwrap() = Some((kind, std::time::Instant::now())); + *self.job.lock().expect("a panic elsewhere poisoned this lock") = Some((kind, std::time::Instant::now())); Job(self) } } @@ -107,8 +108,8 @@ pub struct Job(&'static Stats); impl Drop for Job { fn drop(&mut self) { - if let Some((kind, since)) = self.0.job.lock().unwrap().take() { - let mut jobs = self.0.jobs.lock().unwrap(); + if let Some((kind, since)) = self.0.job.lock().expect("a panic elsewhere poisoned this lock").take() { + let mut jobs = self.0.jobs.lock().expect("a panic elsewhere poisoned this lock"); let e = jobs.entry(kind).or_default(); e.0 += 1; e.1 += since.elapsed().as_secs_f64(); @@ -273,7 +274,7 @@ pub fn render(s: &Sources) -> String { /// The work in progress and done, from `STATS`: jobs, queues and what is in flight (#195). fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_limit: u64) { let now = crate::db::now(); - let job = *st.job.lock().unwrap(); + let job = *st.job.lock().expect("a panic elsewhere poisoned this lock"); gauge(out, "ipx_worker_busy", "1 while the worker is on a job.", &[("", job.is_some() as u8 as f64)]); if let Some((kind, since)) = job { let label = format!("kind=\"{kind}\""); @@ -284,7 +285,7 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li &[(&label, since.elapsed().as_secs_f64())], ); } - let jobs = st.jobs.lock().unwrap().clone(); + let jobs = st.jobs.lock().expect("a panic elsewhere poisoned this lock").clone(); let labels: Vec<String> = jobs.keys().map(|k| format!("kind=\"{k}\"")).collect(); let pairs = |f: fn(&(u64, f64)) -> f64| { labels.iter().zip(jobs.values()).map(|(l, v)| (l.as_str(), f(v))).collect::<Vec<_>>() @@ -359,7 +360,7 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li ); // By person and whose key paid (#278). An account's name is often an email address: /metrics // is on the LAN only, as Prometheus is. - let people = st.people.lock().unwrap().clone(); + let people = st.people.lock().expect("a panic elsewhere poisoned this lock").clone(); let labels: Vec<String> = people.keys().map(|(who, key)| format!("user=\"{}\",key=\"{key}\"", label(who))).collect(); let by = |n: usize| labels.iter().zip(people.values()).map(|(l, v)| (l.as_str(), v[n])).collect::<Vec<_>>(); diff --git a/crates/ipx/src/retention.rs b/crates/ipx/src/retention.rs index 706ae7a..5bcaa22 100644 --- a/crates/ipx/src/retention.rs +++ b/crates/ipx/src/retention.rs @@ -4,7 +4,8 @@ use anyhow::Result; use tracing::Instrument; use crate::config::Config; -use crate::db::{Candidate, Db, now, step}; +use crate::db; +use crate::db::{Candidate, Db}; /// Headroom kept below the quota, as getToQuota()'s `1048576 * 50` did. pub const PAD_BYTES: u64 = 50 * 1024 * 1024; @@ -50,7 +51,7 @@ pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> { // and the whole library be forgotten at once (#293). // Each step is a span of the reap's trace, so a slow one says which part was slow. let missing = if cfg.general.download_dir.is_dir() { - db.missing_files().instrument(step("missing_files")).await? + db.missing_files().instrument(db::step("missing_files")).await? } else { tracing::warn!(dir = %cfg.general.download_dir.display(), "the downloads folder is not there; left every file as it was"); vec![] @@ -63,20 +64,20 @@ pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> { report.reconciled += 1; } - for id in db.prune_abandoned_failures().instrument(step("prune_abandoned_failures")).await? { + for id in db.prune_abandoned_failures().instrument(db::step("prune_abandoned_failures")).await? { tracing::info!(feed = id, "forgot a failing feed nobody subscribes to"); } - let candidates = db.reap_candidates().instrument(step("reap_candidates")).await?; + let candidates = db.reap_candidates().instrument(db::step("reap_candidates")).await?; if cfg.general.max_age_days > 0 { - let cutoff = now() - (cfg.general.max_age_days * 86_400) as i64; + let cutoff = db::now() - (cfg.general.max_age_days * 86_400) as i64; report.aged_out = aged(&candidates, cutoff); for c in &report.aged_out { - report.bytes_freed += remove(db, c, dry_run).instrument(step("remove_aged")).await?; + report.bytes_freed += remove(db, c, dry_run).instrument(db::step("remove_aged")).await?; } if !dry_run { - report.entries_pruned = db.prune_entries(cutoff).instrument(step("prune_entries")).await?; + report.entries_pruned = db.prune_entries(cutoff).instrument(db::step("prune_entries")).await?; } } @@ -88,7 +89,7 @@ pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> { let total: u64 = remaining.iter().map(|c| c.bytes.max(0) as u64).sum(); report.over_quota = pick(&remaining, total, limit); for c in &report.over_quota { - report.bytes_freed += remove(db, c, dry_run).instrument(step("remove_over_quota")).await?; + report.bytes_freed += remove(db, c, dry_run).instrument(db::step("remove_over_quota")).await?; } } diff --git a/crates/ipx/src/signals.rs b/crates/ipx/src/signals.rs index 4e058e8..9aae1ba 100644 --- a/crates/ipx/src/signals.rs +++ b/crates/ipx/src/signals.rs @@ -15,7 +15,8 @@ use tracing::Instrument; use std::sync::atomic::Ordering::Relaxed; use crate::Ctx; -use crate::db::{Ask, Unjudged, step}; +use crate::db; +use crate::db::{Ask, Unjudged}; use crate::feed; use crate::metrics::STATS; @@ -79,8 +80,11 @@ pub fn typesafe_key() -> Option<String> { async fn pass(ctx: &Ctx) -> Result<bool> { let server = server_key(ctx); // People's refused keys are not what is asked with when the server's is. - let skip: Vec<i64> = - if server.is_some() { Vec::new() } else { ctx.signals_refused.lock().unwrap().iter().copied().collect() }; + let skip: Vec<i64> = if server.is_some() { + Vec::new() + } else { + ctx.signals_refused.lock().expect("a panic elsewhere poisoned this lock").iter().copied().collect() + }; let days = ctx.cfg().general.signal_days; let items = ctx.db.unjudged(&skip, BATCH, server.is_some(), days).await?; if items.is_empty() { @@ -110,7 +114,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> { let (group, refused, stopped, server) = (&groups[n], &refused, &stopped, &server); async move { let first = group[0]; - if stopped.load(Relaxed) || refused.lock().unwrap().contains(&first.user_id) { + if stopped.load(Relaxed) || refused.lock().expect("a panic elsewhere poisoned this lock").contains(&first.user_id) { return None; } let key = server.as_deref().or(first.key.as_deref())?; @@ -131,7 +135,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> { *hits_for.entry(item.user_id).or_default() += hits; tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal"); } - ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).instrument(step("record_verdicts")).await?; + ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).instrument(db::step("record_verdicts")).await?; } } // The server's key refused is everyone's: the queue waits for the next pass. @@ -143,7 +147,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> { // A key TypeSafe will not take is not sent the rest of the queue. It is tried // again once its owner saves one (`web::patch_me`) or the daemon restarts. Err(e) if refuses_key(&e) => { - if refused.lock().unwrap().insert(first.user_id) { + if refused.lock().expect("a panic elsewhere poisoned this lock").insert(first.user_id) { tracing::warn!(user = first.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signal key"); } } @@ -173,9 +177,9 @@ async fn pass(ctx: &Ctx) -> Result<bool> { for (user, hits) in hits_for { ctx.out.emit(crate::ipc::Event::SignalHits { user, hits }); } - let refused = refused.into_inner().unwrap(); + let refused = refused.into_inner().expect("a panic elsewhere poisoned this lock"); let some_refused = !refused.is_empty(); - ctx.signals_refused.lock().unwrap().extend(refused); + ctx.signals_refused.lock().expect("a panic elsewhere poisoned this lock").extend(refused); // A pass that only failed would otherwise go straight round again on the same items, and // one TypeSafe refused as malformed would ask again at once, however full. anyhow::Ok(!stopped.into_inner() && (full(&items) || some_refused)) diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index 24854ed..03bbc34 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -21,7 +21,7 @@ use tower::ServiceExt; use tower_http::services::ServeFile; use crate::Ctx; -use crate::db::step; +use crate::db; use crate::ipc::{Command, Event}; use tracing::Instrument; @@ -403,7 +403,7 @@ async fn patch_me( })?; } state.ctx.db.set_typesafe_key(user.id, key.as_deref()).await?; - state.ctx.signals_refused.lock().unwrap().remove(&user.id); + state.ctx.signals_refused.lock().expect("a panic elsewhere poisoned this lock").remove(&user.id); state.ctx.signals.notify_one(); } Ok(StatusCode::NO_CONTENT) @@ -532,7 +532,11 @@ async fn pick_signal_icon( let Some(key) = crate::signals::typesafe_key() else { return Err(ApiError::not_found("picking an icon is not set up on this server")); }; - if !icon_pick_allowed(&mut ICON_PICKS.lock().unwrap(), user.id, std::time::Instant::now()) { + if !icon_pick_allowed( + &mut ICON_PICKS.lock().expect("a panic elsewhere poisoned this lock"), + user.id, + std::time::Instant::now(), + ) { return Err(ApiError::too_many("enough icons picked for one hour; choose one by hand")); } let name = signal_name(&body.name)?; @@ -894,27 +898,28 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>) let cfg = state.ctx.cfg(); // 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(step("catalogue")).await?; + let subs = crate::subscriptions(&state.ctx).instrument(db::step("catalogue")).await?; let mine: std::collections::HashMap<String, crate::db::Sub> = state .ctx .db .subscriptions_for(user.id) - .instrument(step("subscriptions")) + .instrument(db::step("subscriptions")) .await? .into_iter() .map(|s| (s.feed_id.clone(), s)) .collect(); - let counts = state.ctx.db.subscriber_counts().instrument(step("subscriber_counts")).await?; - let pinned = state.ctx.db.pinned_feeds(user.id).instrument(step("pinned")).await?; - let kept = state.ctx.db.kept_feeds(user.id).instrument(step("kept")).await?; - let (hidden, quiet) = state.ctx.db.hidden_from(user.id).instrument(step("hidden_from")).await?; + let counts = state.ctx.db.subscriber_counts().instrument(db::step("subscriber_counts")).await?; + let pinned = state.ctx.db.pinned_feeds(user.id).instrument(db::step("pinned")).await?; + 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 mut listed = state.ctx.db.feed_list(user.id, only).instrument(step("feed_list")).await?; + 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()); - for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id)) { + let asked = subs.iter().filter(|s| only.is_none_or(|o| o == s.id)); + for sub in asked { let (id, feed) = (&sub.id, &sub.cfg); // In a group, what you have not set on the feed comes from your settings on the group, // the same fallback the scanner uses (`Db::subscribers`). @@ -1033,19 +1038,19 @@ async fn rows_for(state: &WebState, user_id: i64, listed: Vec<ipx_directory::Lis .ctx .db .subscriptions_for(user_id) - .instrument(step("subscriptions")) + .instrument(db::step("subscriptions")) .await? .into_iter() .map(|s| s.feed_id) .collect(); let here: std::collections::HashMap<String, String> = crate::subscriptions(&state.ctx) - .instrument(step("catalogue")) + .instrument(db::step("catalogue")) .await? .into_iter() .filter(|s| mine.contains(&s.id)) .filter_map(|s| ipx_directory::store::normalise(&s.cfg.url).map(|u| (u, s.id))) .collect(); - let _matching = step("match").entered(); + let _matching = db::step("match").entered(); Ok(listed .into_iter() .map(|l| { @@ -1219,7 +1224,7 @@ async fn subscribe_popular( let already = existed && state.ctx.db.subscription(user.id, &feed_id).await?.is_some(); state.ctx.db.subscribe(user.id, &feed_id).await?; if !already { - explicit_on_add(&state, user.id, &feed_id, false).await?; + explicit_on_add(&state, user.id, &feed_id, /* allow */ false).await?; } scan_soon(&state, Some(feed_id.clone())).await; Ok(Json(serde_json::json!({ "id": feed_id }))) @@ -1606,20 +1611,20 @@ async fn entry_page( }; let count = async { match signal { - Some(id) => db.signal_count(user_id, id, filter, search).instrument(step("count_in")).await, - None => db.count_of(user_id, feeds, filter, search).instrument(step("count_in")).await, + Some(id) => db.signal_count(user_id, id, filter, search).instrument(db::step("count_in")).await, + None => db.count_of(user_id, feeds, filter, search).instrument(db::step("count_in")).await, } }; // The count beside the list, not after it: each waited on the other, about 40 and 24 ms on the // busiest route (#346). let (mut rows, total) = tokio::try_join!(list, count)?; - step("sanitize").in_scope(|| { + db::step("sanitize").in_scope(|| { let mut sanitizer = feed_sanitizer(); for row in &mut rows { clean_row(&mut sanitizer, row); } }); - crate::deadair::for_reader(&state.ctx, user_id, &mut rows).instrument(step("deadair")).await?; + crate::deadair::for_reader(&state.ctx, user_id, &mut rows).instrument(db::step("deadair")).await?; Ok(Json(EntryPage { total, entries: rows })) } @@ -1771,7 +1776,8 @@ struct FeedPatch { /// looked for in every item of every feed the list covers. fn clean_words(words: Vec<String>) -> Result<Vec<String>, ApiError> { let mut out: Vec<String> = Vec::new(); - for w in words.iter().map(|w| w.trim()).filter(|w| !w.is_empty()) { + let trimmed = words.iter().map(|w| w.trim()).filter(|w| !w.is_empty()); + for w in trimmed { if !out.iter().any(|o| o.eq_ignore_ascii_case(w)) { out.push(w.to_owned()); } @@ -1798,22 +1804,22 @@ async fn patch_feed( ) -> Result<StatusCode, ApiError> { // Pinning is yours alone too, and means nothing for a feed you do not subscribe to. if let Some(on) = body.pinned - && !state.ctx.db.set_pinned(user.id, &id, on).instrument(step("set_pinned")).await? + && !state.ctx.db.set_pinned(user.id, &id, on).instrument(db::step("set_pinned")).await? { return Err(ApiError::not_found("you do not subscribe to that feed")); } if let Some(on) = body.kept - && !state.ctx.db.set_kept(user.id, &id, on).instrument(step("set_kept")).await? + && !state.ctx.db.set_kept(user.id, &id, on).instrument(db::step("set_kept")).await? { return Err(ApiError::not_found("you do not subscribe to that feed")); } if let Some(on) = body.hide_from_all - && !state.ctx.db.set_hide_from_all(user.id, &id, on).instrument(step("set_hide_from_all")).await? + && !state.ctx.db.set_hide_from_all(user.id, &id, on).instrument(db::step("set_hide_from_all")).await? { return Err(ApiError::not_found("you do not subscribe to that feed")); } if let Some(on) = body.hide_from_signals - && !state.ctx.db.set_hide_from_signals(user.id, &id, on).instrument(step("set_hide_from_signals")).await? + && !state.ctx.db.set_hide_from_signals(user.id, &id, on).instrument(db::step("set_hide_from_signals")).await? { return Err(ApiError::not_found("you do not subscribe to that feed")); } @@ -1823,15 +1829,15 @@ async fn patch_feed( return Err(ApiError::bad_request("DeadAir's pause is 200, 300, 400 or 500 ms")); } if let Some(wanted) = body.deadair { - state.ctx.db.set_deadair(user.id, &id, wanted).instrument(step("set_deadair")).await?; + state.ctx.db.set_deadair(user.id, &id, wanted).instrument(db::step("set_deadair")).await?; state.ctx.deadair.notify_one(); } if let Some(p) = body.deadair_pause { - state.ctx.db.set_deadair_pause(user.id, &id, p).instrument(step("set_deadair_pause")).await?; + state.ctx.db.set_deadair_pause(user.id, &id, p).instrument(db::step("set_deadair_pause")).await?; } // What one person wants -- which items, whether to fetch them, how many at a time -- // is theirs. It goes on their subscription and nobody else sees the change. - if let Some(mut mine) = state.ctx.db.subscription(user.id, &id).instrument(step("subscription")).await? { + if let Some(mut mine) = state.ctx.db.subscription(user.id, &id).instrument(db::step("subscription")).await? { let mut touched = false; if let Some(v) = body.keywords.clone() { mine.keywords = Some(v.into_iter().filter(|k| !k.trim().is_empty()).collect()); @@ -1850,10 +1856,10 @@ async fn patch_feed( touched = true; } if touched { - state.ctx.db.set_subscription(user.id, &mine).instrument(step("set_subscription")).await?; + state.ctx.db.set_subscription(user.id, &mine).instrument(db::step("set_subscription")).await?; } if let Some(v) = body.blocked.clone() { - state.ctx.db.set_blocklist(user.id, &id, &clean_words(v)?).instrument(step("set_blocklist")).await?; + state.ctx.db.set_blocklist(user.id, &id, &clean_words(v)?).instrument(db::step("set_blocklist")).await?; } } @@ -1874,7 +1880,7 @@ async fn patch_feed( } // Marking a feed private is for those who read it: anyone could keep anyone's feed, or a // whole OPML's, out of the directory, and only an admin could undo it (#333). - if !user.is_admin && state.ctx.db.subscription(user.id, &id).instrument(step("subscription")).await?.is_none() { + if !user.is_admin && state.ctx.db.subscription(user.id, &id).instrument(db::step("subscription")).await?.is_none() { return Err(ApiError::not_found("you do not subscribe to that feed")); } let mut cfg = state.ctx.edit_cfg().await; @@ -1882,7 +1888,7 @@ async fn patch_feed( // Derived feeds have no config entry. Editing one is the moment it earns a real // entry: promote it, so the config holds your decisions and nothing else. if !cfg.feeds.contains_key(&id) { - let subs = crate::subscriptions(&state.ctx).instrument(step("subscriptions")).await?; + let subs = crate::subscriptions(&state.ctx).instrument(db::step("subscriptions")).await?; let found = subs.iter().find(|s| s.id == id).ok_or_else(|| ApiError::bad_request(format!("no feed with id {id:?}")))?; cfg.feeds.insert(id.clone(), found.cfg.clone()); @@ -1926,7 +1932,7 @@ async fn patch_feed( if url_changed { // Refreshing a rotated auth token is the common case; entries and download history // are keyed by feed id, so they survive the change. - state.ctx.db.clear_validators(&id).instrument(step("clear_validators")).await?; + state.ctx.db.clear_validators(&id).instrument(db::step("clear_validators")).await?; } Ok(StatusCode::NO_CONTENT) } @@ -2137,7 +2143,7 @@ fn changed_feed(ev: &Event) -> Option<&str> { /// Audio, served by ServeFile so Range requests work and the player can seek. async fn media(State(state): State<WebState>, Path(id): Path<i64>, user: crate::db::User, req: Request) -> Response { - serve_media(state, id, user, req, false).await + serve_media(state, id, user, req, /* piece */ false).await } /// DeadAir's playlist of a file (#181): the original's frames kept by this person's pause @@ -2150,7 +2156,8 @@ async fn media_deadair(State(state): State<WebState>, Path(id): Path<i64>, user: if let Err(e) = subscribed(&state, &user, &enc.feed_id).await { return e.into_response(); } - let plan = match crate::deadair::plan_for(&state.ctx, user.id, &enc.feed_id, id).instrument(step("plan")).await { + let plan = match crate::deadair::plan_for(&state.ctx, user.id, &enc.feed_id, id).instrument(db::step("plan")).await + { Ok(Some(p)) => p, Ok(None) => return (StatusCode::NOT_FOUND, "DeadAir has nothing to cut in this file for you").into_response(), Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(), @@ -2178,7 +2185,8 @@ async fn media_save(State(state): State<WebState>, Path(id): Path<i64>, user: cr let Some(path) = enc.path else { return (StatusCode::NOT_FOUND, "not downloaded").into_response(); }; - let plan = match crate::deadair::plan_for(&state.ctx, user.id, &enc.feed_id, id).instrument(step("plan")).await { + let plan = match crate::deadair::plan_for(&state.ctx, user.id, &enc.feed_id, id).instrument(db::step("plan")).await + { Ok(Some(p)) => p, Ok(None) => return (StatusCode::NOT_FOUND, "DeadAir has nothing to cut in this file for you").into_response(), Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(), @@ -2212,7 +2220,7 @@ async fn media_piece( user: crate::db::User, req: Request, ) -> Response { - serve_media(state, id, user, req, true).await + serve_media(state, id, user, req, /* piece */ true).await } async fn serve_media(state: WebState, id: i64, user: crate::db::User, req: Request, piece: bool) -> Response { @@ -2294,7 +2302,7 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon // Only what a feed here or the directory names: anything else would make ipx fetch any // address it is given. The Directory shows podcasts nobody here subscribes to (#273). if !crate::directory::listed_art(&q.u) - && !state.ctx.db.names_image(&q.u).instrument(step("names_image")).await.unwrap_or(false) + && !state.ctx.db.names_image(&q.u).instrument(db::step("names_image")).await.unwrap_or(false) { return (StatusCode::NOT_FOUND, "no feed names that artwork").into_response(); } @@ -2323,10 +2331,10 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon let kind = image_kind(&said, &body).ok_or_else(|| anyhow::anyhow!("not an image: {said}"))?; anyhow::Ok((kind, body)) }; - match got.instrument(step("art_fetch")).await { + match got.instrument(db::step("art_fetch")).await { Ok((kind, body)) => { if keep { - let (span, file, kind, body) = (step("art_keep"), file.clone(), kind.clone(), body.clone()); + let (span, file, kind, body) = (db::step("art_keep"), file.clone(), kind.clone(), body.clone()); match tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::write(&file, &kind, &body))) .await { @@ -2368,7 +2376,7 @@ impl Drop for ArtFetchDone<'_> { /// Artwork kept on disk, read off the async threads: the cache is on FUSE, and four calls a /// hit, fifty images a page, held a worker each, 1 s at p90 for a hit (#345). async fn kept_art(file: &std::path::Path) -> Option<(String, Vec<u8>)> { - let (span, file) = (step("art_cached"), file.to_owned()); + let (span, file) = (db::step("art_cached"), file.to_owned()); tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::read(&file))).await.ok().flatten() }