rays/ipx
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases 20 Wiki Activity

Follow the Rust style guide's smaller rules (#353)

Browse Source 
Measured against /src/styleguide/rustguide.md:
- Each use statement imports from one module, and free functions are
  called through theirs (db::step, db::now, jsonwebtoken::decode,
  crate::feed::redact, unix::signal) rather than imported bare.
- unsafe_code is forbidden workspace-wide. The one unsafe was a test's
  set_var, which is unsound while other tests' threads read the
  environment; the test now reads PATH, which every run has.
- Locks say why they may panic: expect("a panic elsewhere poisoned this
  lock") in place of 25 unwraps, and unique_slug's endless search says
  it cannot end.
- No combinator chains in for headers: each is a named variable first.
- #[must_use] on the 27 functions clippy's must_use_candidate names.
- A literal bool passed where its meaning is not plain gets its name
  (/* keep_open */ true, /* piece */ false, ...).

No behaviour changes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-24 01:07:09 +00:00
parent 59c7f5f213
commit 41affc8d8e
25 changed files with 202 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
crates/ipx/Cargo.toml
View File

@@ -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

20
crates/ipx/src/access.rs
View File

@@ -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()) }

3
crates/ipx/src/auth.rs
View File

@@ -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.

9
crates/ipx/src/config.rs
View File

@@ -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");
}
}

1
crates/ipx/src/db.rs
View File

@@ -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> {

16
crates/ipx/src/deadair.rs
View File

@@ -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| {

3
crates/ipx/src/directory.rs
View File

@@ -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! {

3
crates/ipx/src/download.rs
View File

@@ -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(' ');
}

15
crates/ipx/src/ipc.rs
View File

@@ -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) => {

68
crates/ipx/src/main.rs
View File

@@ -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}");
}

17
crates/ipx/src/metrics.rs
View File

@@ -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<_>>();

17
crates/ipx/src/retention.rs
View File

@@ -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?;
}
}

20
crates/ipx/src/signals.rs
View File

@@ -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))

88
crates/ipx/src/web.rs
View File

@@ -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()
}