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

Import type names instead of writing their paths (#355)

Browse Source 
The style guide imports types and uses them unqualified. 222 places
wrote a path such as std::collections::HashMap or std::time::Duration
inline; each file now imports the type once, one use a module, and
names it bare. Done with a script, file by file, only where the name is
unambiguous there:
- one path for the name throughout the file, and not a name the file
  already binds otherwise (its own types, other imports, a function's
  local use);
- not a name generic enough to mislead bare (Error, Result, Value,
  Mutex, Client, Request, Response, Command, Event, Context, Map, Ok);
- only code a normal build compiles, so no import is unused outside
  tests or names a crate only tests depend on.

193 paths stay, those cases above. No behaviour change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-24 01:19:10 +00:00
parent d33d83043d
commit f22c1993b1
25 changed files with 307 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
crates/directory/src/apple.rs
View File

@@ -13,6 +13,7 @@ use serde::Deserialize;
use tracing::Instrument;
use crate::{Shared, SnapshotEntry};
use std::collections::{BTreeSet, HashSet};
/// What the charts are filed under, as a server's snapshot is under its token's name.
pub const INSTANCE: &str = "apple-podcasts";
@@ -83,7 +84,7 @@ async fn sync(dir: &Shared, client: &reqwest::Client, country: &str) -> Result<(
}
tokio::time::sleep(PAUSE).await;
}
let mut seen = std::collections::HashSet::new();
let mut seen = HashSet::new();
ids.retain(|id| seen.insert(id.clone()));
anyhow::ensure!(!ids.is_empty(), "no chart had a show in it");
@@ -150,9 +151,8 @@ fn differ(theirs: &BTreeMap<String, Vec<String>>, ours: &[(&str, &[&str])]) -> (
let flat = |c: &str, subs: &mut dyn Iterator<Item = String>| {
std::iter::once(c.to_string()).chain(subs.map(|s| format!("{c} / {s}"))).collect::<Vec<_>>()
};
let theirs: std::collections::BTreeSet<String> =
theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect();
let ours: std::collections::BTreeSet<String> =
let theirs: BTreeSet<String> = theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect();
let ours: BTreeSet<String> =
ours.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().map(|s| s.to_string()))).collect();
(theirs.difference(&ours).cloned().collect(), ours.difference(&theirs).cloned().collect())
}

10
crates/directory/src/config.rs
View File

@@ -4,6 +4,8 @@
use anyhow::{Context, Result, bail};
use serde::Deserialize;
use std::collections::HashSet;
use std::fs::File;
use std::path::Path;
/// `directory.toml`. See docs/directory.md.
@@ -87,7 +89,7 @@ impl Config {
/// instances' snapshots together.
pub fn parse(text: &str) -> Result<Self> {
let cfg: Config = toml::from_str(text)?;
let mut seen = std::collections::HashSet::new();
let mut seen = HashSet::new();
for t in &cfg.tokens {
if t.name.trim().is_empty() {
bail!("a token has an empty name");
@@ -111,7 +113,7 @@ impl Config {
/// The tokens as the server checks them. A hash that is not 64 hex digits, or the same
/// hash under two names, is an error: which name a request is would be ambiguous.
pub fn tokens(&self) -> Result<Vec<Token>> {
let mut seen = std::collections::HashSet::new();
let mut seen = HashSet::new();
self.tokens
.iter()
.map(|t| {
@@ -148,9 +150,7 @@ pub fn new_token() -> String {
use std::io::Read;
let mut bytes = [0u8; 32];
// Falling back to the clock would make a guessable token. Better to fail.
std::fs::File::open("/dev/urandom")
.and_then(|mut f| f.read_exact(&mut bytes))
.expect("no source of randomness for a token");
File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut bytes)).expect("no source of randomness for a token");
format!("ipxd_{}", bytes.iter().map(|b| format!("{b:02x}")).collect::<String>())
}

26
crates/directory/src/lib.rs
View File

@@ -26,7 +26,13 @@ use axum::{Extension, Json, Router};
use serde::{Deserialize, Serialize};
use config::{Access, Token};
use reqwest::redirect::Policy;
use std::path::PathBuf;
use store::{Refreshed, Store};
use tokio::sync::Notify;
use tower_http::compression::CompressionLayer;
use tower_http::decompression::RequestDecompressionLayer;
use tracing::Instrument;
/// A feed as the directory lists it.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -134,7 +140,7 @@ pub struct Directory {
tokens: RwLock<Vec<Token>>,
client: reqwest::Client,
/// Wakes the refresher, so a feed a snapshot brings in is fetched now, not in a minute.
wake: tokio::sync::Notify,
wake: Notify,
/// The public pages' copy of the list, and when it was read.
pages_list: tokio::sync::Mutex<Option<(Instant, Arc<Vec<Listing>>)>>,
}
@@ -155,9 +161,9 @@ impl Directory {
) -> Result<Shared, reqwest::Error> {
// No redirects: ipx_feed::fetch follows them itself, so it sees each hop.
let client = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(10))
.connect_timeout(Duration::from_secs(10))
.read_timeout(ipx_feed::FEED_SILENCE)
.redirect(reqwest::redirect::Policy::none())
.redirect(Policy::none())
.user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION")))
// The feeds are named by whoever holds a write token, and redirect where their
// publishers like: nothing on this network is fetched unless listed (#329).
@@ -167,7 +173,7 @@ impl Directory {
store,
tokens: RwLock::new(tokens),
client,
wake: tokio::sync::Notify::new(),
wake: Notify::new(),
pages_list: Default::default(),
}))
}
@@ -239,14 +245,14 @@ pub async fn refresher(dir: Shared) {
if !due.is_empty() {
// A round is a trace, each feed's check a span in it (#213).
let round = tracing::info_span!("refresh", feeds = due.len());
tracing::Instrument::instrument(
Instrument::instrument(
async {
// A few at a time: in turn, two publishers timing out held a round
// for a minute and more (#227).
futures_util::StreamExt::for_each_concurrent(futures_util::stream::iter(&due), 5, |d| {
let check = tracing::info_span!("feed", feed = %d.id);
async {
if let Err(e) = tracing::Instrument::instrument(dir.refresh(d), check).await {
if let Err(e) = Instrument::instrument(dir.refresh(d), check).await {
tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check");
}
}
@@ -272,7 +278,7 @@ pub async fn refresher(dir: Shared) {
/// Reads the config file again whenever it changes, checked each minute. A file that fails to
/// load leaves the tokens as they were, so a typo locks nobody out.
pub async fn reloader(dir: Shared, path: std::path::PathBuf) {
pub async fn reloader(dir: Shared, path: PathBuf) {
let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok();
let mut last = mtime(&path);
loop {
@@ -321,9 +327,9 @@ pub fn router(dir: Shared) -> Router {
// Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of
// JSON, most of a Directory page's time (#253). Inside the access log, which records what
// was sent.
.layer(tower_http::compression::CompressionLayer::new().br(true).zstd(true).gzip(true))
.layer(CompressionLayer::new().br(true).zstd(true).gzip(true))
// And takes a request's body gzipped: ipx sends its snapshot so, about 100 KB as JSON.
.layer(tower_http::decompression::RequestDecompressionLayer::new().gzip(true))
.layer(RequestDecompressionLayer::new().gzip(true))
.layer(axum::middleware::from_fn(access_log))
.with_state(dir)
}
@@ -368,7 +374,7 @@ async fn access_log(req: Request, next: Next) -> Response {
use tracing_opentelemetry::OpenTelemetrySpanExt;
let _ = span.set_parent(parent);
}
let resp = tracing::Instrument::instrument(next.run(req), span.clone()).await;
let resp = Instrument::instrument(next.run(req), span.clone()).await;
span.record("http.response.status_code", resp.status().as_u16());
let _in = span.enter();
let who = resp.extensions().get::<Who>().map_or("", |w| w.0.as_str()).to_owned();

24
crates/directory/src/main.rs
View File

@@ -4,6 +4,11 @@ use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use ipx_directory::store::Store;
use ipx_directory::{Directory, config};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::net::TcpListener;
use tokio::signal::unix::SignalKind;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::prelude::*;
#[derive(Parser)]
@@ -12,7 +17,7 @@ struct Cli {
/// The config file: where to listen, the database, and the tokens it accepts. Otherwise
/// DIRECTORY_CONFIG, or else /config/directory.toml.
#[arg(long)]
config: Option<std::path::PathBuf>,
config: Option<PathBuf>,
#[command(subcommand)]
cmd: Option<Cmd>,
}
@@ -62,7 +67,7 @@ async fn main() -> Result<()> {
let port = cfg.bind.rsplit(':').next().unwrap_or("8100");
let resp = reqwest::Client::new()
.get(format!("http://127.0.0.1:{port}/healthz"))
.timeout(std::time::Duration::from_secs(4))
.timeout(Duration::from_secs(4))
.send()
.await
.context("the directory does not answer")?;
@@ -72,11 +77,10 @@ async fn main() -> Result<()> {
}
}
async fn serve(path: &std::path::Path) -> Result<()> {
async fn serve(path: &Path) -> Result<()> {
// One JSON object a line for Loki when DIRECTORY_LOG_FORMAT=json, as ipx does; text otherwise.
let json = std::env::var("DIRECTORY_LOG_FORMAT").is_ok_and(|f| f.eq_ignore_ascii_case("json"));
let filter =
|| tracing_subscriber::EnvFilter::try_from_env("DIRECTORY_LOG").unwrap_or_else(|_| "ipx_directory=info".into());
let filter = || EnvFilter::try_from_env("DIRECTORY_LOG").unwrap_or_else(|_| "ipx_directory=info".into());
// Traces to Tempo when OTEL_EXPORTER_OTLP_ENDPOINT is set, joining ipx's (#213).
let otel = ipx_directory::otel::provider("ipx-directory")?;
tracing_subscriber::registry()
@@ -92,7 +96,7 @@ async fn serve(path: &std::path::Path) -> Result<()> {
use opentelemetry::trace::TracerProvider;
tracing_opentelemetry::layer()
.with_tracer(p.tracer("ipx-directory"))
.with_filter(tracing_subscriber::EnvFilter::new("ipx_directory=info,ipx_feed=info"))
.with_filter(EnvFilter::new("ipx_directory=info,ipx_feed=info"))
}))
.init();
@@ -106,11 +110,10 @@ async fn serve(path: &std::path::Path) -> Result<()> {
let dir = Directory::new(store, vec![], ipx_feed::guard::Allowed::parse(&cfg.fetch_internal)?)?;
let dropped = dir.set_tokens(tokens).await?;
let listener =
tokio::net::TcpListener::bind(&cfg.bind).await.with_context(|| format!("listening on {}", cfg.bind))?;
let listener = TcpListener::bind(&cfg.bind).await.with_context(|| format!("listening on {}", cfg.bind))?;
tracing::info!(version = env!("CARGO_PKG_VERSION"), bind = %cfg.bind, tokens = cfg.tokens.len(), dropped, "directory started");
if !cfg.metrics_bind.is_empty() {
let metrics = tokio::net::TcpListener::bind(&cfg.metrics_bind)
let metrics = TcpListener::bind(&cfg.metrics_bind)
.await
.with_context(|| format!("listening for metrics on {}", cfg.metrics_bind))?;
tracing::info!(bind = %cfg.metrics_bind, "serving metrics for Prometheus at /metrics");
@@ -125,8 +128,7 @@ async fn serve(path: &std::path::Path) -> Result<()> {
tokio::spawn(ipx_directory::reloader(dir.clone(), path.to_owned()));
axum::serve(listener, ipx_directory::router(dir))
.with_graceful_shutdown(async {
let mut term =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("SIGTERM handler");
let mut term = tokio::signal::unix::signal(SignalKind::terminate()).expect("SIGTERM handler");
tokio::select! {
_ = term.recv() => {}
_ = tokio::signal::ctrl_c() => {}

14
crates/directory/src/metrics.rs
View File

@@ -6,7 +6,11 @@
//! mimalloc gives freed pages back, so resident memory follows what is in use: a spike that
//! does not come down again is a leak. Not jemalloc, whose upstream was archived in 2025.
use axum::Router;
use std::sync::atomic::Ordering::Relaxed;
use std::time::Instant;
use tokio::net::TcpListener;
use tower_http::compression::CompressionLayer;
#[global_allocator]
static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -25,7 +29,7 @@ pub fn process() -> (Option<f64>, Option<f64>) {
/// The directory's /metrics, in Prometheus's text format: memory, open files, and the feeds it
/// has fetched. Nothing from the database, so a scrape costs nothing and answers while it is down.
fn render(started: std::time::Instant) -> String {
fn render(started: Instant) -> String {
let mut out = String::new();
let mut gauge = |name: &str, kind: &str, help: &str, labels: &str, v: f64| {
out.push_str(&format!("# HELP {name} {help}\n# TYPE {name} {kind}\n{name}{labels} {v}\n"));
@@ -70,9 +74,9 @@ fn render(started: std::time::Instant) -> String {
/// Serves /metrics on a listener of its own: the public port is behind the tunnel with no
/// Access in front, and these are for the LAN.
pub async fn serve(listener: tokio::net::TcpListener) {
let started = std::time::Instant::now();
let app = axum::Router::new()
pub async fn serve(listener: TcpListener) {
let started = Instant::now();
let app = Router::new()
.route(
"/metrics",
axum::routing::get(move || async move {
@@ -80,7 +84,7 @@ pub async fn serve(listener: tokio::net::TcpListener) {
}),
)
// Prometheus asks for gzip (#253).
.layer(tower_http::compression::CompressionLayer::new().gzip(true));
.layer(CompressionLayer::new().gzip(true));
if let Err(e) = axum::serve(listener, app).await {
tracing::error!(error = %e, "metrics stopped");
}

4
crates/directory/src/otel.rs
View File

@@ -2,6 +2,8 @@
//! caller and out to a callee, so one trace follows a request from the browser through ipx-web
//! and ipx into the directory.
use axum::http::HeaderMap;
/// Traces over OTLP, to Tempo for one, when OTEL_EXPORTER_OTLP_ENDPOINT names a collector
/// (`http://host:4318`: the exporter speaks OTLP over HTTP and adds `/v1/traces`). The exporter
/// reads that and the other `OTEL_` variables itself.
@@ -24,7 +26,7 @@ pub fn provider(service: &'static str) -> anyhow::Result<Option<opentelemetry_sd
/// header can be set: a position saved by sendBeacon, and the playlist Safari's player fetches
/// itself (#192).
#[must_use]
pub fn remote_parent(headers: &axum::http::HeaderMap, query: Option<&str>) -> Option<opentelemetry::Context> {
pub fn remote_parent(headers: &HeaderMap, query: Option<&str>) -> Option<opentelemetry::Context> {
use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId, TraceState};
let v = match headers.get("traceparent") {
Some(h) => h.to_str().ok()?,

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

@@ -16,6 +16,7 @@ use axum::response::{IntoResponse, Response};
use axum::routing::get;
use crate::{Listing, Shared};
use url::form_urlencoded::Serializer;
pub fn router() -> Router<Shared> {
Router::new()
@@ -140,7 +141,7 @@ async fn listed(dir: &Shared, title: &str, here: &str, f: &Filter, pick: impl Fn
// The query a link keeps, with one of its keys changed.
let link = |kind: Option<&str>, sub: Option<&str>| {
let mut q = url::form_urlencoded::Serializer::new(String::new());
let mut q = Serializer::new(String::new());
if let Some(v) = f.q.as_deref() {
q.append_pair("q", v);
}

23
crates/directory/src/store.rs
View File

@@ -6,6 +6,11 @@ use anyhow::Result;
use sea_orm::{ConnectionTrait, DatabaseConnection, Statement, TransactionTrait, Value};
use crate::{Blocked, FeedPage, Instance, Item, Listing, SnapshotEntry, SnapshotResult};
use sea_orm::{ConnectOptions, Database, QueryResult};
use std::cmp::Reverse;
use std::collections::{BTreeMap, HashSet};
use std::time::SystemTime;
use url::Url;
/// A feed's items as its page shows them, and as many as ipx's own Directory page showed.
pub const ITEMS: usize = 20;
@@ -42,14 +47,14 @@ pub enum Refreshed {
/// Unix seconds now.
#[must_use]
pub fn now() -> i64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64)
SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64)
}
/// The address as every instance would write it: scheme and host lower-cased, a default port
/// and a fragment dropped. None for anything that is not http(s).
#[must_use]
pub fn normalise(url: &str) -> Option<String> {
let mut u = url::Url::parse(url.trim()).ok()?;
let mut u = Url::parse(url.trim()).ok()?;
if !matches!(u.scheme(), "http" | "https") || u.host_str().is_none() {
return None;
}
@@ -146,7 +151,7 @@ impl Store {
let sep = if url.contains('?') { '&' } else { '?' };
url = format!("{url}{sep}options=-c%20client_min_messages%3Dwarning");
}
let mut opts = sea_orm::ConnectOptions::new(url.clone());
let mut opts = ConnectOptions::new(url.clone());
opts.sqlx_logging(false);
// An in-memory SQLite database is one per connection: a pool of them would be several.
if url.contains(":memory:") {
@@ -157,7 +162,7 @@ impl Store {
if keep_open && url.starts_with("postgres") {
opts.max_connections(10).min_connections(10);
}
let db = sea_orm::Database::connect(opts).await?;
let db = Database::connect(opts).await?;
for sql in SCHEMA {
db.execute_unprepared(sql).await?;
}
@@ -177,7 +182,7 @@ impl Store {
Ok(self.db.execute_raw(self.stmt(sql, values)).await?.rows_affected())
}
async fn rows(&self, sql: &str, values: Vec<Value>) -> Result<Vec<sea_orm::QueryResult>> {
async fn rows(&self, sql: &str, values: Vec<Value>) -> Result<Vec<QueryResult>> {
Ok(self.db.query_all_raw(self.stmt(sql, values)).await?)
}
@@ -209,14 +214,14 @@ impl Store {
/// blocked feed is left out and counted; the rest are added to the feeds, new ones to be
/// fetched by the refresher. Feeds no instance sends any more go.
pub async fn snapshot(&self, instance: &str, entries: &[SnapshotEntry]) -> Result<SnapshotResult> {
let blocked: std::collections::HashSet<String> = self
let blocked: HashSet<String> = self
.rows("SELECT id FROM blocked", vec![])
.await?
.iter()
.map(|r| r.try_get("", "id"))
.collect::<Result<_, _>>()?;
let mut result = SnapshotResult::default();
let mut keep = std::collections::BTreeMap::new();
let mut keep = BTreeMap::new();
for e in entries {
let Some(url) = normalise(&e.url) else {
result.refused_invalid += 1;
@@ -307,7 +312,7 @@ impl Store {
f.podcast, CAST(sum(s.subscribers) AS BIGINT) AS subscribers
FROM feeds f JOIN snapshots s ON s.feed_id = f.id WHERE f.items IS NOT NULL";
fn listing(r: &sea_orm::QueryResult) -> Result<Listing> {
fn listing(r: &QueryResult) -> Result<Listing> {
Ok(Listing {
id: r.try_get("", "id")?,
url: r.try_get("", "url")?,
@@ -483,7 +488,7 @@ impl Store {
.flat_map(|e| &e.enclosures)
.any(|e| e.mime.as_deref().is_some_and(|m| m.starts_with("audio/") || m.starts_with("video/")));
let mut entries = parsed.entries;
entries.sort_by_key(|e| std::cmp::Reverse(e.published));
entries.sort_by_key(|e| Reverse(e.published));
// What an item is, never its files: their addresses are where a key would be.
let items: Vec<Item> = entries
.into_iter()

3
crates/feed/src/guard.rs
View File

@@ -14,6 +14,7 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use anyhow::{Context as _, Result};
use http::Extensions;
use hyper_util::client::legacy::connect::{Connection, HttpInfo};
type BoxError = Box<dyn std::error::Error + Send + Sync>;
@@ -153,7 +154,7 @@ where
let allowed = self.allowed.clone();
Box::pin(async move {
let conn = connecting.await.map_err(Into::into)?;
let mut extras = http::Extensions::new();
let mut extras = Extensions::new();
conn.connected().get_extras(&mut extras);
// Refused when the peer is unknown too: a connection this cannot check is one it
// cannot vouch for.

37
crates/feed/src/lib.rs
View File

@@ -4,8 +4,13 @@
pub mod guard;
use anyhow::{Context, Result, anyhow};
use chrono::{DateTime, NaiveDateTime};
use reqwest::StatusCode;
use reqwest::header::HeaderName;
use reqwest::header::{ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED};
use std::collections::{HashMap, HashSet};
use std::sync::atomic::AtomicU64;
use std::time::Duration;
/// A feed as `parse` reads it, RSS or Atom alike: the channel's own details and its items.
#[derive(Debug, Default)]
@@ -73,20 +78,20 @@ pub enum Fetched {
/// with no limit one hung server held every scan for as long as it did. Dreamwidth answered 504
/// after 60-67 s for a day, and each scan took 70-80 s instead of 15 (#108). A feed is small;
/// downloads, which are not, have no such limit.
pub const FEED_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
pub const FEED_TIMEOUT: Duration = Duration::from_secs(30);
/// The longest a feed's server may send nothing, for the clients that fetch feeds to set: one
/// that took the connection and never answered ran the whole 30 s, every scan it was in (#230).
/// No feed that answered took over 8 s in a day of traces; one that keeps sending, however
/// slowly, still has until FEED_TIMEOUT.
pub const FEED_SILENCE: std::time::Duration = std::time::Duration::from_secs(15);
pub const FEED_SILENCE: Duration = Duration::from_secs(15);
/// Feeds fetched and the bytes of their bodies, since the process started, for /metrics to set
/// beside memory (#224): a body is read whole, so a night of large feeds is a night of large
/// allocations.
pub static FETCHES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
pub static FETCHES: AtomicU64 = AtomicU64::new(0);
/// See `FETCHES`.
pub static FETCHED_BYTES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
pub static FETCHED_BYTES: AtomicU64 = AtomicU64::new(0);
/// Conditional GET, following redirects itself: `client` must follow none (`feed_client`), so
/// each hop is seen. The second value is where the feed now is when every hop said so for good
@@ -204,8 +209,7 @@ pub async fn fetch(
// The original surfaced 401/407 specially; the code is enough for a UI to switch on.
return Err(anyhow!("HTTP {status}"));
}
let header =
|h: reqwest::header::HeaderName| resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned);
let header = |h: HeaderName| resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned);
let etag = header(ETAG);
let last_modified = header(LAST_MODIFIED);
let bytes = read_capped(resp).await.context("reading body")?;
@@ -257,7 +261,7 @@ fn from_wp_posts(posts: &[serde_json::Value]) -> ParsedFeed {
link: non_empty(p["link"].as_str()),
published: p["date_gmt"]
.as_str()
.and_then(|d| chrono::NaiveDateTime::parse_from_str(d, "%Y-%m-%dT%H:%M:%S").ok())
.and_then(|d| NaiveDateTime::parse_from_str(d, "%Y-%m-%dT%H:%M:%S").ok())
.map(|d| d.and_utc().timestamp()),
description: non_empty(p["content"]["rendered"].as_str()),
image: non_empty(p["jetpack_featured_media_url"].as_str()),
@@ -647,7 +651,7 @@ fn alternate_feed_link(bytes: &[u8]) -> Option<String> {
/// feed that moved (#102); cnn.com is one.
pub async fn find_feed(client: &reqwest::Client, url: &str) -> Result<String> {
let read = |u: String| async move {
let resp = client.get(&u).timeout(std::time::Duration::from_secs(30)).send().await.context("connecting")?;
let resp = client.get(&u).timeout(Duration::from_secs(30)).send().await.context("connecting")?;
anyhow::ensure!(resp.status().is_success(), "HTTP {}", resp.status());
let base = resp.url().clone();
anyhow::Ok((base, read_capped(resp).await.context("reading")?))
@@ -675,7 +679,7 @@ pub async fn find_feed(client: &reqwest::Client, url: &str) -> Result<String> {
/// `/favicon.ico`. None if neither is there.
#[tracing::instrument(skip_all, fields(site = site))]
pub async fn site_icon(client: &reqwest::Client, site: &str) -> Option<String> {
let timeout = std::time::Duration::from_secs(20);
let timeout = Duration::from_secs(20);
let resp = client.get(site).timeout(timeout).send().await.ok()?;
// Relative to where the page ended up, not where it was asked for: a site that redirects
// to /en/ would otherwise have its icon looked for in the wrong place.
@@ -702,11 +706,7 @@ const MAX_ART_HOSTS: usize = 8;
/// address it does not serve on https stays, for /api/art to fetch (#90). Four of the five
/// hosts the catalogue had on http served the same image on https; The Secret Cabal's CDN
/// presents another name's certificate (#110).
pub async fn prefer_https(
client: &reqwest::Client,
url: &str,
known: &mut std::collections::HashMap<String, bool>,
) -> String {
pub async fn prefer_https(client: &reqwest::Client, url: &str, known: &mut HashMap<String, bool>) -> String {
let Some(rest) = url.strip_prefix("http://") else { return url.to_owned() };
let host = rest.split('/').next().unwrap_or("").to_owned();
let secure = format!("https://{rest}");
@@ -728,7 +728,7 @@ pub async fn prefer_https(
/// Whether `url` answers with an image. A site with no favicon often answers 200 with its home
/// page, which is not an icon.
pub async fn is_image(client: &reqwest::Client, url: &str) -> bool {
let timeout = std::time::Duration::from_secs(20);
let timeout = Duration::from_secs(20);
let Ok(resp) = client.get(url).timeout(timeout).send().await else { return false };
resp.status().is_success()
&& resp
@@ -1037,7 +1037,7 @@ fn non_empty(s: Option<&str>) -> Option<String> {
/// that embeds the file it encloses lists the same file twice: Rands in Repose's "The Promotion
/// Paradox" was downloaded twice and offered two play buttons for one mp3. The first stays.
fn drop_player_repeats(encs: &mut Vec<Enclosure>) {
let mut seen = std::collections::HashSet::new();
let mut seen = HashSet::new();
encs.retain(|e| seen.insert(same_file_key(&e.url)));
}
@@ -1238,10 +1238,7 @@ fn parse_duration(s: &str) -> Option<i64> {
/// RSS pubDate is RFC 2822; some feeds ship RFC 3339 instead.
fn parse_date(s: &str) -> Option<i64> {
let s = s.trim();
chrono::DateTime::parse_from_rfc2822(s)
.or_else(|_| chrono::DateTime::parse_from_rfc3339(s))
.ok()
.map(|d| d.timestamp())
DateTime::parse_from_rfc2822(s).or_else(|_| DateTime::parse_from_rfc3339(s)).ok().map(|d| d.timestamp())
}
/// Apple's podcast categories, each with its subcategories

11
crates/ipx/src/art.rs
View File

@@ -2,8 +2,11 @@
//! of each publisher's server for every visit, and still there when a publisher is not.
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::{LazyLock, Mutex};
use std::time::SystemTime;
use std::time::{Duration, Instant};
pub fn dir() -> PathBuf {
@@ -16,7 +19,7 @@ pub fn dir() -> PathBuf {
// one in 10^12. Move to a SHA if either ever matters.
pub fn path(url: &str) -> PathBuf {
use std::hash::{Hash, Hasher};
let mut h = std::collections::hash_map::DefaultHasher::new();
let mut h = DefaultHasher::new();
url.hash(&mut h);
dir().join(format!("{:016x}", h.finish()))
}
@@ -25,8 +28,8 @@ pub fn path(url: &str) -> PathBuf {
pub fn read(file: &Path) -> Option<(String, Vec<u8>)> {
let kind = std::fs::read_to_string(file.with_extension("type")).ok()?;
let body = std::fs::read(file).ok()?;
if let Ok(f) = std::fs::File::options().write(true).open(file) {
let _ = f.set_modified(std::time::SystemTime::now());
if let Ok(f) = File::options().write(true).open(file) {
let _ = f.set_modified(SystemTime::now());
}
Some((kind, body))
}
@@ -75,7 +78,7 @@ pub fn fail(url: &str, why: &str) {
/// Returns how many went.
pub fn trim(dir: &Path, limit: u64) -> usize {
let Ok(entries) = std::fs::read_dir(dir) else { return 0 };
let mut kept: Vec<(std::time::SystemTime, u64, PathBuf)> = entries
let mut kept: Vec<(SystemTime, u64, PathBuf)> = entries
.flatten()
.map(|e| e.path())
.filter(|p| p.extension().is_none())

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

@@ -6,6 +6,9 @@ use anyhow::{Result, bail};
use argon2::Argon2;
use argon2::password_hash::phc::PasswordHash;
use argon2::password_hash::{PasswordHasher, PasswordVerifier};
use std::fs::File;
use std::sync::{Arc, LazyLock};
use tokio::sync::Semaphore;
/// Argon2id with the crate's defaults, which are the OWASP-recommended parameters. The
/// salt is generated per password by the hasher itself.
@@ -32,17 +35,15 @@ pub fn verify_password(password: &str, stored: &str) -> bool {
/// How many password checks run at once: each is tens of milliseconds of CPU, and forty wrong
/// passwords at a time, run on the async workers, made every other request wait 4s (#137).
/// Half the cores, so a flood of sign-ins waits on itself and the rest of the server has the rest.
static CHECKS: std::sync::LazyLock<std::sync::Arc<tokio::sync::Semaphore>> = std::sync::LazyLock::new(|| {
std::sync::Arc::new(tokio::sync::Semaphore::new(
std::thread::available_parallelism().map_or(1, |n| (n.get() / 2).max(1)),
))
static CHECKS: LazyLock<Arc<Semaphore>> = LazyLock::new(|| {
Arc::new(Semaphore::new(std::thread::available_parallelism().map_or(1, |n| (n.get() / 2).max(1))))
});
/// What a name that is not an account is checked against, so that refusing it takes as long as
/// refusing a wrong password does. Refused without a check, it came back 31ms sooner, and the
/// time told anyone which names are accounts here (#138).
static DECOY: std::sync::LazyLock<String> =
std::sync::LazyLock::new(|| hash_password("no account here has this password").expect("hashing a fixed password"));
static DECOY: LazyLock<String> =
LazyLock::new(|| hash_password("no account here has this password").expect("hashing a fixed password"));
/// A sign-in's password against the account's hash, or against `DECOY` when there is no account
/// or it has no password: false either way, in the same time. Off the async workers, and a few at
@@ -91,7 +92,7 @@ pub fn api_token_hash(token: &str) -> String {
fn getrandom(buf: &mut [u8]) -> std::io::Result<()> {
use std::io::Read;
std::fs::File::open("/dev/urandom")?.read_exact(buf)
File::open("/dev/urandom")?.read_exact(buf)
}
/// A username taken from a proxy header. Cloudflare sends an email address; the local part

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

@@ -3,7 +3,9 @@
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fs::Permissions;
use std::path::{Path, PathBuf};
use url::Url;
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
pub struct Config {
@@ -319,7 +321,7 @@ impl Feed {
}
fn same_host(a: &str, b: &str) -> bool {
let host = |u: &str| url::Url::parse(u).ok().and_then(|u| u.host_str().map(str::to_ascii_lowercase));
let host = |u: &str| Url::parse(u).ok().and_then(|u| u.host_str().map(str::to_ascii_lowercase));
host(a).is_some_and(|h| host(b).as_ref() == Some(&h))
}
@@ -456,7 +458,7 @@ fn write_private(path: &Path, text: &str) -> Result<()> {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
std::fs::set_permissions(path, Permissions::from_mode(0o600))?;
}
Ok(())
}

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

@@ -7,15 +7,19 @@ use crate::entity::{
};
use anyhow::{Context, Result};
use sea_orm::sea_query::{Expr, Func};
use sea_orm::sqlx::sqlite::SqliteSynchronous;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set,
Statement,
};
use sea_orm::{ConnectOptions, Database, DatabaseConnection};
use std::path::Path;
use std::time::SystemTime;
use tracing::Instrument;
/// A pool of connections, where there was one connection behind a global lock.
pub struct Db {
orm: sea_orm::DatabaseConnection,
orm: DatabaseConnection,
/// A test's database file, removed when the test is done with it.
#[cfg(test)]
tmp: Option<std::path::PathBuf>,
@@ -46,7 +50,7 @@ impl Drop for Db {
/// here, as the old migrate() did for SQLite, or sea-orm-migration once there are several. A
/// database from before 0.7 took its last columns from that migrate(), so it upgrades through
/// a 0.7 release first.
async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
async fn create_missing(orm: &DatabaseConnection) -> Result<()> {
use crate::entity::*;
use sea_orm::{ConnectionTrait, Schema};
let schema = Schema::new(orm.get_database_backend());
@@ -241,15 +245,15 @@ fn is_postgres(location: &str) -> bool {
/// sqlx's defaults for SQLite are what ipx wants, foreign keys on and a five-second wait for a
/// lock, which is what rusqlite was set to, but for one.
async fn connect(location: &str, keep_open: bool) -> Result<sea_orm::DatabaseConnection> {
let mut opts = sea_orm::ConnectOptions::new(url_for(location));
async fn connect(location: &str, keep_open: bool) -> Result<DatabaseConnection> {
let mut opts = ConnectOptions::new(url_for(location));
opts.sqlx_logging(false);
// NORMAL, SQLite's own advice for WAL: synced at checkpoints, not after every commit. FULL,
// the default, waited for the disk after nearly every statement, and a feed's items are
// written a row at a time: 1.3s a feed on Tower's /tmp, where one synchronous write takes
// 52ms, and 39s to subscribe an admin to 1,500 feeds (#135). A crash of ipx loses nothing;
// only a power cut can lose the last transactions, and it never corrupts the file.
opts.map_sqlx_sqlite_opts(|o| o.synchronous(sea_orm::sqlx::sqlite::SqliteSynchronous::Normal));
opts.map_sqlx_sqlite_opts(|o| o.synchronous(SqliteSynchronous::Normal));
// The same on Postgres, for ipx's own sessions: a commit waited for the WAL to reach the
// disk, 12.8ms on the databases project's server, and at about fifty commits a feed a scan
// took most of a second a feed (#140). A crash of the Postgres server can lose the last
@@ -271,7 +275,7 @@ async fn connect(location: &str, keep_open: bool) -> Result<sea_orm::DatabaseCon
if keep_open && is_postgres(location) {
opts.min_connections(20);
}
sea_orm::Database::connect(opts).await.with_context(|| format!("opening {}", redact(location)))
Database::connect(opts).await.with_context(|| format!("opening {}", redact(location)))
}
/// The URL `connect` hands sqlx for a location: a SQLite file, created if missing, or a Postgres
@@ -532,7 +536,7 @@ impl Db {
let one = || sea_orm::Value::from(only.map(str::to_owned));
// A step a query: the feed list's trace was 290 ms with nothing under it.
let counts = |of: &'static str, sql: &'static str, args: Vec<sea_orm::Value>| async move {
tracing::Instrument::instrument(self.rows(sql, args), step(of))
Instrument::instrument(self.rows(sql, args), step(of))
.await?
.iter()
.map(|r| Ok((r.try_get::<String>("", "feed_id")?, r.try_get::<i64>("", "n")?)))
@@ -568,7 +572,7 @@ impl Db {
),
vec![user_id.into(), one()],
);
let blocked = tracing::Instrument::instrument(
let blocked = Instrument::instrument(
blocklists::Entity::find().filter(blocklists::Column::UserId.eq(user_id)).all(&self.orm),
step("blocked"),
);
@@ -576,7 +580,7 @@ impl Db {
if let Some(id) = only {
rows = rows.filter(feeds::Column::Id.eq(id));
}
let rows = tracing::Instrument::instrument(rows.all(&self.orm), step("feeds"));
let rows = Instrument::instrument(rows.all(&self.orm), step("feeds"));
let (entries, downloaded, unread, blocked, rows) = tokio::try_join!(
entries,
downloaded,
@@ -819,7 +823,7 @@ impl Db {
/// Returns true when this enclosure URL is new. False means we have downloaded it
/// before, or deliberately reaped it -- either way it is not fetched again.
pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> {
pub async fn mark_downloaded(&self, url: &str, path: &Path, bytes: u64) -> Result<()> {
self.exec(
"UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3,
downloaded_at = $4, last_error = NULL WHERE url = $1",
@@ -1002,7 +1006,7 @@ impl Db {
.await?
.into_iter()
.filter_map(|e| Some((e.id, e.path?)))
.filter(|(_, p)| !std::path::Path::new(p).exists())
.filter(|(_, p)| !Path::new(p).exists())
.collect())
}
@@ -3086,7 +3090,7 @@ impl Db {
}
pub fn now() -> i64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0)
SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0)
}
#[cfg(test)]

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

@@ -26,6 +26,12 @@ use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use crate::Ctx;
use axum::body::Bytes;
use std::collections::{HashSet, VecDeque};
use std::io::SeekFrom;
use std::process::Stdio;
use std::time::{Duration, Instant};
use tokio::fs::File;
/// Quieter than this for longer than `MIN_SILENCE` is a silence: what the measurements on the
/// issue used. A podcast's room tone sits well below it and speech well above.
@@ -44,7 +50,7 @@ pub const DEFAULT_PAUSE: i64 = 300;
const MAX_PIECE: f64 = 10.0;
/// How long the maker waits when there is nothing to do: a download or a feed turned on is
/// picked up this long after, at most.
const IDLE: std::time::Duration = std::time::Duration::from_secs(120);
const IDLE: Duration = Duration::from_secs(120);
/// A span of the original, in milliseconds, that DeadAir leaves out.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -81,28 +87,28 @@ pub fn id3_len(head: &[u8]) -> u64 {
pub async fn kept(
path: &Path,
plan: &Plan,
) -> std::io::Result<(u64, impl futures_util::Stream<Item = std::io::Result<axum::body::Bytes>> + use<>)> {
) -> std::io::Result<(u64, impl futures_util::Stream<Item = std::io::Result<Bytes>> + use<>)> {
use tokio::io::{AsyncReadExt, AsyncSeekExt};
let mut file = tokio::fs::File::open(path).await?;
let mut file = File::open(path).await?;
let mut head = [0u8; 10];
let n = file.read(&mut head).await?;
// Only up to the first frame kept: past the tag can be the Xing frame, which gives the
// original's length.
let first = plan.pieces.first().map_or(0, |p| p.0);
let tag = id3_len(&head[..n]).min(first);
let ranges: std::collections::VecDeque<(u64, u64)> =
let ranges: VecDeque<(u64, u64)> =
std::iter::once((0, tag)).chain(plan.pieces.iter().map(|&(o, n, _)| (o, n))).filter(|r| r.1 > 0).collect();
let len = ranges.iter().map(|r| r.1).sum();
let body = futures_util::stream::try_unfold((file, ranges), |(mut f, mut left)| async move {
let Some((off, n)) = left.pop_front() else { return Ok(None) };
let mut buf = vec![0; n.min(1 << 16) as usize];
f.seek(std::io::SeekFrom::Start(off)).await?;
f.seek(SeekFrom::Start(off)).await?;
f.read_exact(&mut buf).await?;
let got = buf.len() as u64;
if got < n {
left.push_front((off + got, n - got));
}
Ok(Some((axum::body::Bytes::from(buf), (f, left))))
Ok(Some((Bytes::from(buf), (f, left))))
});
Ok((len, body))
}
@@ -371,7 +377,7 @@ pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
}
// Files ffmpeg failed on, passed over until the next start. Not stored: a failure recorded
// as the file having nothing to cut was never tried again, even once a fix was deployed (#174).
let mut failed = std::collections::HashSet::new();
let mut failed = HashSet::new();
loop {
// Dropping the pass on a signal kills the ffmpeg it is running (`kill_on_drop`).
let done = tokio::select! {
@@ -394,7 +400,7 @@ pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
async fn have_ffmpeg() -> bool {
tokio::process::Command::new("ffmpeg")
.arg("-version")
.stdout(std::process::Stdio::null())
.stdout(Stdio::null())
.status()
.await
.is_ok_and(|s| s.success())
@@ -404,7 +410,7 @@ async fn have_ffmpeg() -> bool {
/// shortened has already heard through gets none: finished episodes are rarely played again.
// shortcut: those stay on the list and are looked at again each pass, one query each; record
// them as passed over if a large library makes the pass slow.
async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result<bool> {
async fn next(ctx: &Ctx, failed: &mut HashSet<i64>) -> Result<bool> {
// Nobody with DeadAir on for anything, which is most passes: nothing more to ask the database.
use std::sync::atomic::Ordering::Relaxed;
let st = &crate::metrics::STATS;
@@ -433,7 +439,7 @@ async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result<
continue;
}
let src = crate::download::playable(Path::new(&t.path));
let started = std::time::Instant::now();
let started = Instant::now();
// A trace of its own for each file looked at, not for the passes that find nothing to do,
// which are most of them.
let span = tracing::info_span!("deadair", feed = t.feed_id, enclosure = t.id, cuts = tracing::field::Empty);
@@ -594,7 +600,7 @@ async fn run_out(prog: &str, args: &[&str]) -> Result<String> {
fn command(prog: &str) -> tokio::process::Command {
let mut c = tokio::process::Command::new(prog);
c.kill_on_drop(true).stdin(std::process::Stdio::null());
c.kill_on_drop(true).stdin(Stdio::null());
c
}

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

@@ -8,9 +8,13 @@ use ipx_directory::{FeedPage, Listing, Me, SnapshotEntry, SnapshotResult};
use serde::de::DeserializeOwned;
use crate::{Ctx, Sub, config, feed};
use std::fmt::{Display, Formatter};
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant};
use tracing::Instrument;
/// How often the snapshot goes when nothing has woken the sender sooner.
const SHARE_EVERY: std::time::Duration = std::time::Duration::from_secs(30 * 60);
const SHARE_EVERY: Duration = Duration::from_secs(30 * 60);
/// What went wrong asking the directory, in words for an admin.
#[derive(Debug)]
@@ -23,8 +27,8 @@ pub enum DirError {
Unreachable(String),
}
impl std::fmt::Display for DirError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
impl Display for DirError {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
DirError::NotConfigured => {
write!(f, "No directory is set up on this server. An admin sets one in Admin, Directory.")
@@ -69,11 +73,8 @@ async fn call<T: DeserializeOwned>(
body: Option<&[SnapshotEntry]>,
) -> Result<T, DirError> {
let (base, token) = target(&ctx.cfg()).ok_or(DirError::NotConfigured)?;
let mut req = ctx
.local_client
.request(method, format!("{base}{path}"))
.bearer_auth(token)
.timeout(std::time::Duration::from_secs(20));
let mut req =
ctx.local_client.request(method, format!("{base}{path}")).bearer_auth(token).timeout(Duration::from_secs(20));
// The directory's spans join this request's trace (#213).
if let Some(tp) = ipx_directory::otel::traceparent() {
req = req.header("traceparent", tp);
@@ -139,7 +140,7 @@ pub async fn page(ctx: &Ctx, id: &str) -> Result<FeedPage, DirError> {
///
/// shortcut: in memory, one address a listed feed, forgotten on a restart until the Directory is
/// next read, which the page does when it loads; a table if that ever shows as initials.
static LISTED_ART: std::sync::LazyLock<std::sync::Mutex<HashSet<String>>> = std::sync::LazyLock::new(Default::default);
static LISTED_ART: LazyLock<std::sync::Mutex<HashSet<String>>> = LazyLock::new(Default::default);
fn note_art<'a>(listings: impl Iterator<Item = &'a Listing>) {
let mut art = LISTED_ART.lock().unwrap_or_else(|e| e.into_inner());
@@ -240,10 +241,10 @@ pub async fn share(ctx: &Ctx, entries: &[SnapshotEntry]) -> Result<SnapshotResul
}
/// Sends the snapshot every half hour while sharing is on, and at once when the settings change.
pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
// What went last, where, and when: a scan wakes this every half minute or so, and the same
// 1,500 feeds went each time, the directory rewriting its rows for nothing (#211).
let mut last: Option<(Vec<u8>, std::time::Instant)> = None;
let mut last: Option<(Vec<u8>, Instant)> = None;
loop {
let cfg = ctx.cfg();
if cfg.general.directory_share && configured(&cfg) {
@@ -254,9 +255,9 @@ pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receive
if forced || resend(last.as_ref(), &key) {
let span =
tracing::info_span!(target: "ipx::directory", "directory_share", feeds = entries.len());
let got = tracing::Instrument::instrument(share(&ctx, &entries), span).await;
let got = Instrument::instrument(share(&ctx, &entries), span).await;
if got.is_ok() {
last = Some((key, std::time::Instant::now()));
last = Some((key, Instant::now()));
}
Some(got)
} else {
@@ -294,7 +295,7 @@ pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receive
/// Whether to send the snapshot `key` stands for: one that changed, or went to another directory
/// or with another token, goes now; the same again only once SHARE_EVERY has passed, so a
/// directory that lost it, or dropped this server's counts, has it back within the half hour.
fn resend(last: Option<&(Vec<u8>, std::time::Instant)>, key: &[u8]) -> bool {
fn resend(last: Option<&(Vec<u8>, Instant)>, key: &[u8]) -> bool {
last.is_none_or(|(k, at)| k != key || at.elapsed() >= SHARE_EVERY)
}

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

@@ -6,6 +6,9 @@ use std::path::{Path, PathBuf};
use tokio::io::AsyncWriteExt;
use crate::config::{Config, Feed as FeedCfg, Organize};
use chrono::Local;
use tokio::fs::File;
use url::Url;
/// Forbidden characters that were separating words: they become "-" so the words stay
/// apart. The original's stringCleaning() deleted them, turning "Show | Series" into
@@ -83,7 +86,7 @@ pub fn filename_for(url: &str, content_disposition: Option<&str>) -> String {
if let Some(name) = content_disposition.and_then(disposition_filename) {
return sanitize(&name);
}
let from_url = url::Url::parse(url)
let from_url = Url::parse(url)
.ok()
.and_then(|u| u.path_segments().and_then(|mut s| s.rfind(|p| !p.is_empty())).map(str::to_owned))
.map(|seg| percent_encoding::percent_decode_str(&seg).decode_utf8_lossy().into_owned())
@@ -151,7 +154,7 @@ pub async fn download(
tokio::fs::create_dir_all(&tmp_dir).await.with_context(|| format!("creating {}", tmp_dir.display()))?;
let tmp = tmp_dir.join(&name);
let mut file = tokio::fs::File::create(&tmp).await.with_context(|| format!("creating {}", tmp.display()))?;
let mut file = File::create(&tmp).await.with_context(|| format!("creating {}", tmp.display()))?;
let mut done: u64 = 0;
let mut stream = resp.bytes_stream();
let _active = crate::metrics::busy(&crate::metrics::STATS.downloads_active);
@@ -246,7 +249,7 @@ async fn sniff(path: &Path) -> Result<Sniffed> {
async fn read_head(path: &Path, n: usize) -> Result<Vec<u8>> {
use tokio::io::AsyncReadExt;
let mut f = tokio::fs::File::open(path).await?;
let mut f = File::open(path).await?;
let mut buf = vec![0u8; n];
let read = f.read(&mut buf).await?;
buf.truncate(read);
@@ -277,7 +280,7 @@ fn unique_path(dir: &Path, name: &str) -> PathBuf {
/// letting the sanitizer eat the separator.
pub fn folder_for(cfg: &Config, id: &str, feed_cfg: &FeedCfg, title: Option<&str>) -> String {
match cfg.general.organize {
Organize::Date => chrono::Local::now().format("%m-%d-%Y").to_string(),
Organize::Date => Local::now().format("%m-%d-%Y").to_string(),
Organize::Feed => {
let raw = feed_cfg.folder.as_deref().or(title).filter(|s| !s.trim().is_empty()).unwrap_or(id);
raw.split('/')

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

@@ -2,7 +2,12 @@
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::{broadcast, mpsc};
@@ -124,7 +129,7 @@ impl Event {
/// Whether a page signed in as this person is sent it. Everyone had every event, so a key in
/// a paid feed's path, which `redacted` keeps, reached people who do not read the feed (#325).
/// An admin has them all, for the log.
pub fn is_for(&self, user: &crate::db::User, mine: &std::collections::HashSet<String>) -> bool {
pub fn is_for(&self, user: &crate::db::User, mine: &HashSet<String>) -> bool {
if user.is_admin {
return true;
}
@@ -350,8 +355,7 @@ pub async fn daemon_is_live(path: &Path) -> bool {
/// the meantime. The answer goes to the client that asked and no one else: broadcast, it ended
/// any `ipx fetch` that was watching a scan, since `status` is a terminal event.
/// A future, since reading the counts is a database query.
pub type StatusFn =
std::sync::Arc<dyn Fn(Command) -> std::pin::Pin<Box<dyn std::future::Future<Output = Event> + Send>> + Send + Sync>;
pub type StatusFn = Arc<dyn Fn(Command) -> Pin<Box<dyn Future<Output = Event> + Send>> + Send + Sync>;
/// Accepts connections, feeding commands to `cmds` and events from `events` back out.
pub async fn serve(
@@ -455,9 +459,7 @@ pub async fn reload(path: &Path) -> Result<()> {
}
anyhow::bail!("the daemon closed the connection without answering")
};
tokio::time::timeout(std::time::Duration::from_secs(10), answer)
.await
.context("the daemon did not answer in 10s")?
tokio::time::timeout(Duration::from_secs(10), answer).await.context("the daemon did not answer in 10s")?
}
/// Sends one command to a running daemon and prints the events it produces.

6
crates/ipx/src/logbuf.rs
View File

@@ -4,7 +4,9 @@
//! to read. Capturing inside the tracing pipeline works the same either way.
use std::collections::VecDeque;
use std::fmt::Debug;
use std::sync::{LazyLock, Mutex};
use tracing::Subscriber;
use tracing::field::{Field, Visit};
use tracing_subscriber::Layer;
use tracing_subscriber::layer::Context;
@@ -67,7 +69,7 @@ pub fn since(after: u64, limit: usize) -> (Vec<LogLine>, u64) {
/// A tracing layer that mirrors every event into the ring.
pub struct RingLayer;
impl<S: tracing::Subscriber> Layer<S> for RingLayer {
impl<S: Subscriber> Layer<S> for RingLayer {
fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
let mut v = Collect::default();
event.record(&mut v);
@@ -102,7 +104,7 @@ impl Collect {
}
impl Visit for Collect {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
fn record_debug(&mut self, field: &Field, value: &dyn Debug) {
self.add(field, format!("{value:?}"));
}
// Numbers and bools reach record_debug through the trait's defaults, which prints them the

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

@@ -18,12 +18,29 @@ mod torrent;
mod web;
use anyhow::{Context, Result};
use chrono::DateTime;
use clap::{Parser, Subcommand};
use ipc::{Command as Cmd, Emitter, Event};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fs::OpenOptions;
use std::future::Future;
use std::io::IsTerminal;
use std::ops::{Deref, DerefMut};
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicI64};
use std::time::{Duration, Instant};
use tokio::runtime::Handle;
use tokio::sync::{MutexGuard, Notify, OnceCell, Semaphore};
use tokio::sync::{broadcast, mpsc, watch};
use tokio::task::JoinHandle;
use tracing::Subscriber;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::fmt::format::JsonFields;
use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
use tracing_subscriber::registry::LookupSpan;
use url::Url;
#[derive(Parser)]
#[command(name = "ipx", version, about = "A headless podcatcher")]
@@ -122,7 +139,7 @@ enum UserCmd {
/// `admin-password` beside config.toml, readable by the daemon's user alone. Every install began
/// as admin / ipodderx, published in the README, and nothing made anyone change it (#321).
/// Returns the password and where it can be found, for the log, which never shows it.
fn first_password(config_dir: &std::path::Path, from_env: Option<String>) -> Result<(String, String)> {
fn first_password(config_dir: &Path, from_env: Option<String>) -> Result<(String, String)> {
if let Some(p) = from_env.filter(|p| !p.is_empty()) {
return Ok((p, "with the password in IPX_ADMIN_PASSWORD".into()));
}
@@ -133,12 +150,12 @@ fn first_password(config_dir: &std::path::Path, from_env: Option<String>) -> Res
}
/// Writes a secret only its owner may read: created 0600, never readable for a moment between.
fn write_private(file: &std::path::Path, text: &str) -> Result<()> {
fn write_private(file: &Path, text: &str) -> Result<()> {
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
std::fs::create_dir_all(file.parent().unwrap_or(std::path::Path::new(".")))?;
std::fs::create_dir_all(file.parent().unwrap_or(Path::new(".")))?;
let _ = std::fs::remove_file(file);
let mut f = std::fs::OpenOptions::new()
let mut f = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
@@ -153,7 +170,7 @@ pub struct Ctx {
/// Swapped wholesale when the web UI rewrites config.toml, so a running daemon picks
/// up feed changes without a restart. Callers take a snapshot; no guard is ever held
/// across an await.
pub cfg: std::sync::RwLock<std::sync::Arc<config::Config>>,
pub cfg: std::sync::RwLock<Arc<config::Config>>,
pub db: db::Db,
/// For anything a feed or a person names: refuses internal addresses (#320).
pub client: reqwest::Client,
@@ -166,9 +183,9 @@ pub struct Ctx {
pub out: Emitter,
/// Started on first use: a BitTorrent session binds ports and starts a DHT, which is
/// rude to do for a config that has never seen a torrent.
pub torrents: tokio::sync::OnceCell<torrent::Torrents>,
pub torrents: OnceCell<torrent::Torrents>,
/// Caps how many torrents run at once when they are detached.
pub torrent_slots: Arc<tokio::sync::Semaphore>,
pub torrent_slots: Arc<Semaphore>,
/// Needed because a subscribed OPML rewrites the feed list as it syncs.
pub config_path: PathBuf,
/// Run torrents off the command worker. A torrent takes minutes to fetch metadata and
@@ -179,7 +196,7 @@ pub struct Ctx {
/// Feeds whose host could not be looked up for the moment, and when to try them again: a
/// failure that says nothing about the feed, kept off it and out of the database (#143).
/// Forgotten on a restart, which only makes the retry sooner.
pub retry: std::sync::Mutex<std::collections::HashMap<String, i64>>,
pub retry: std::sync::Mutex<HashMap<String, i64>>,
/// Held from taking a copy of the config to storing it (`edit_cfg`). Each change stores the
/// whole catalogue from its copy, and two at once stored over each other: a feed added while
/// another add spent a minute fetching was gone once that one stored (#153).
@@ -189,37 +206,37 @@ pub struct Ctx {
pub cfg_write: tokio::sync::Mutex<()>,
/// Wakes DeadAir's maker when a show is turned on, so its copies start at once instead of
/// at the next pass (#171).
pub deadair: tokio::sync::Notify,
pub deadair: Notify,
/// Wakes the Signals judge when there are new items or a Signal is made or changed (#254).
pub signals: tokio::sync::Notify,
pub signals: Notify,
/// People whose TypeSafe key was refused, passed over until they save another.
pub signals_refused: std::sync::Mutex<std::collections::HashSet<i64>>,
pub signals_refused: std::sync::Mutex<HashSet<i64>>,
/// When the Directory was last cleared of dead feeds (`clean_directory`), 0 for not since
/// starting.
pub directory_cleaned: std::sync::atomic::AtomicI64,
pub directory_cleaned: AtomicI64,
/// Sends the snapshot to the directory now, as when its settings change (#205).
pub directory_wake: tokio::sync::Notify,
pub directory_wake: Notify,
/// When the snapshot last went and what came of it, for the admin page.
pub directory_shared: std::sync::Mutex<Option<directory::Shared>>,
/// Send the snapshot even if it has not changed: after an unblock or a dropped snapshot the
/// directory needs it again, though nothing here differs (#206).
pub directory_resend: std::sync::atomic::AtomicBool,
pub directory_resend: AtomicBool,
}
/// A copy of the config to change, holding off every other change until `store_cfg` keeps it.
pub struct CfgEdit<'a> {
cfg: config::Config,
_held: tokio::sync::MutexGuard<'a, ()>,
_held: MutexGuard<'a, ()>,
}
impl std::ops::Deref for CfgEdit<'_> {
impl Deref for CfgEdit<'_> {
type Target = config::Config;
fn deref(&self) -> &config::Config {
&self.cfg
}
}
impl std::ops::DerefMut for CfgEdit<'_> {
impl DerefMut for CfgEdit<'_> {
fn deref_mut(&mut self) -> &mut config::Config {
&mut self.cfg
}
@@ -261,7 +278,7 @@ impl Ctx {
self.db.set_feed_error(id, url, msg).await
}
pub fn cfg(&self) -> std::sync::Arc<config::Config> {
pub fn cfg(&self) -> Arc<config::Config> {
self.cfg.read().expect("a panic elsewhere poisoned this lock").clone()
}
@@ -282,7 +299,7 @@ impl Ctx {
}
fn set_cfg(&self, cfg: config::Config) {
*self.cfg.write().expect("a panic elsewhere poisoned this lock") = std::sync::Arc::new(cfg);
*self.cfg.write().expect("a panic elsewhere poisoned this lock") = Arc::new(cfg);
tracing::info!("config reloaded");
// A feed added, removed, made private or given a category, or the directory's own
// settings changed: the directory hears of it now, not at the next half hour (#205).
@@ -312,25 +329,23 @@ async fn main() -> Result<()> {
// Two filters, deliberately different. stderr follows IPX_LOG; the in-app buffer
// keeps debug as well, so the log view can show protocol traffic and routine
// skips that would be noise on a terminal. IPX_UI_LOG overrides it.
let stderr_filter =
|| tracing_subscriber::EnvFilter::try_from_env("IPX_LOG").unwrap_or_else(|_| "ipx=info".into());
let stderr_filter = || EnvFilter::try_from_env("IPX_LOG").unwrap_or_else(|_| "ipx=info".into());
// One JSON object a line for Loki (#91), with each event's fields as its own; text
// otherwise, for someone reading a terminal.
let json = std::env::var("IPX_LOG_FORMAT").is_ok_and(|f| f.eq_ignore_ascii_case("json"));
let ui_filter =
tracing_subscriber::EnvFilter::try_from_env("IPX_UI_LOG").unwrap_or_else(|_| "ipx=debug".into());
let ui_filter = EnvFilter::try_from_env("IPX_UI_LOG").unwrap_or_else(|_| "ipx=debug".into());
tracing_subscriber::registry()
.with((!json).then(|| {
tracing_subscriber::fmt::layer()
.with_writer(std::io::stderr)
// Colour for a terminal only: in docker logs and Loki the escapes are noise
// every query has to strip (#88).
.with_ansi(std::io::IsTerminal::is_terminal(&std::io::stderr()))
.with_ansi(IsTerminal::is_terminal(&std::io::stderr()))
.with_filter(stderr_filter())
}))
.with(json.then(|| {
tracing_subscriber::fmt::layer()
.fmt_fields(tracing_subscriber::fmt::format::JsonFields::new())
.fmt_fields(JsonFields::new())
.event_format(WithTrace(
tracing_subscriber::fmt::format()
.json()
@@ -346,7 +361,7 @@ async fn main() -> Result<()> {
use opentelemetry::trace::TracerProvider;
tracing_opentelemetry::layer()
.with_tracer(p.tracer("ipx"))
.with_filter(tracing_subscriber::EnvFilter::new("ipx=info,ipx_feed=info"))
.with_filter(EnvFilter::new("ipx=info,ipx_feed=info"))
}))
.init();
}
@@ -390,24 +405,24 @@ async fn main() -> Result<()> {
let (events, _) = broadcast::channel(1024);
let guard = feed::guard::Guard::new(feed::guard::Allowed::parse(&cfg.general.fetch_internal)?);
let ctx = Arc::new(Ctx {
cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)),
cfg: std::sync::RwLock::new(Arc::new(cfg)),
db,
client: reqwest::Client::builder()
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
// Connecting only, so it bounds a download's start, not a long download (#108).
.connect_timeout(std::time::Duration::from_secs(10))
.connect_timeout(Duration::from_secs(10))
// Two minutes without a byte is a stalled download, however long it is: unbounded, one
// held the worker, and every scan behind it, for good, while the daemon read healthy (#152).
.read_timeout(std::time::Duration::from_secs(120))
.read_timeout(Duration::from_secs(120))
.connector_layer(guard.clone())
.build()?,
local_client: reqwest::Client::builder()
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
.connect_timeout(std::time::Duration::from_secs(10))
.connect_timeout(Duration::from_secs(10))
.build()?,
feed_client: reqwest::Client::builder()
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
.connect_timeout(std::time::Duration::from_secs(10))
.connect_timeout(Duration::from_secs(10))
.read_timeout(feed::FEED_SILENCE)
.redirect(reqwest::redirect::Policy::none())
.connector_layer(guard)
@@ -417,8 +432,8 @@ async fn main() -> Result<()> {
} else {
Emitter::terminal()
},
torrents: tokio::sync::OnceCell::new(),
torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)),
torrents: OnceCell::new(),
torrent_slots: Arc::new(Semaphore::new(2)),
config_path: config_path.clone(),
detach_torrents: is_daemon,
retry: Default::default(),
@@ -463,15 +478,15 @@ async fn main() -> Result<()> {
/// the object it writes. A line outside any traced span, or with no trace exporter, is unchanged.
struct WithTrace<F>(F);
impl<S, N, F> tracing_subscriber::fmt::FormatEvent<S, N> for WithTrace<F>
impl<S, N, F> FormatEvent<S, N> for WithTrace<F>
where
F: tracing_subscriber::fmt::FormatEvent<S, N>,
S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
N: for<'w> tracing_subscriber::fmt::FormatFields<'w> + 'static,
F: FormatEvent<S, N>,
S: Subscriber + for<'a> LookupSpan<'a>,
N: for<'w> FormatFields<'w> + 'static,
{
fn format_event(
&self,
ctx: &tracing_subscriber::fmt::FmtContext<'_, S, N>,
ctx: &FmtContext<'_, S, N>,
mut w: tracing_subscriber::fmt::format::Writer<'_>,
ev: &tracing::Event<'_>,
) -> std::fmt::Result {
@@ -533,7 +548,7 @@ async fn user_cmd(ctx: &Arc<Ctx>, cmd: UserCmd) -> Result<()> {
for u in users {
let added = u
.created
.and_then(|t| chrono::DateTime::from_timestamp(t, 0))
.and_then(|t| DateTime::from_timestamp(t, 0))
.map_or("?".into(), |d| d.format("%Y-%m-%d").to_string());
let seen = u.last_login.map_or("never signed in".into(), |t| format!("signed in {}", ago(Some(t))));
println!(
@@ -652,7 +667,7 @@ async fn daemon(
// A database with nobody in it cannot be signed into.
if ctx.db.users().await?.is_empty() {
let dir = ctx.config_path.parent().unwrap_or(std::path::Path::new("."));
let dir = ctx.config_path.parent().unwrap_or(Path::new("."));
let (password, told) = first_password(dir, std::env::var("IPX_ADMIN_PASSWORD").ok())?;
ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?;
tracing::warn!(
@@ -732,7 +747,7 @@ async fn daemon(
{
let (ctx, mut stop) = (ctx.clone(), rx_stop.clone());
tokio::spawn(async move {
let mut every = tokio::time::interval(std::time::Duration::from_secs(30));
let mut every = tokio::time::interval(Duration::from_secs(30));
loop {
tokio::select! {
_ = every.tick() => { status(&ctx).await; }
@@ -766,7 +781,7 @@ async fn daemon(
if *stop.borrow() {
break;
}
let wait = if std::mem::take(&mut first) { std::time::Duration::ZERO } else { until_next_scan(&ctx).await };
let wait = if std::mem::take(&mut first) { Duration::ZERO } else { until_next_scan(&ctx).await };
tokio::select! {
biased;
_ = stop.changed() => break,
@@ -805,11 +820,11 @@ async fn daemon(
/// Starts the web UI when it is switched on, minting and saving a token if there is none.
async fn start_web(
ctx: &Arc<Ctx>,
config_path: &std::path::Path,
config_path: &Path,
web_addr: Option<String>,
cmds: &mpsc::Sender<Cmd>,
events: &broadcast::Sender<Event>,
) -> Result<Option<tokio::task::JoinHandle<()>>> {
) -> Result<Option<JoinHandle<()>>> {
let cfg = ctx.cfg();
let enabled = cfg.web.enabled || web_addr.is_some();
if !enabled {
@@ -866,8 +881,8 @@ async fn start_web(
cmds: cmds.clone(),
events: events.clone(),
playing: playing.clone(),
runtime: tokio::runtime::Handle::current(),
started: std::time::Instant::now(),
runtime: Handle::current(),
started: Instant::now(),
};
let at = metrics::spawn(&cfg.web.metrics_bind, sources)?;
tracing::info!(%at, "serving metrics for Prometheus at /metrics");
@@ -991,7 +1006,7 @@ async fn add_one(
// Slugs must be unique across derived feeds too, or a new feed can collide with one
// an OPML already introduced.
let mut taken: std::collections::BTreeMap<String, config::Feed> =
let mut taken: BTreeMap<String, config::Feed> =
subscriptions(ctx).await?.into_iter().map(|s| (s.id, s.cfg)).collect();
// A removed feed keeps its rows, so its id is only free again for the same feed: re-adding
// it gets its history back, and a different feed does not inherit someone else's.
@@ -1007,7 +1022,7 @@ async fn add_one(
/// Host plus last path segment, for naming a feed we could not read.
fn url_stem(url: &str) -> String {
url::Url::parse(url).ok().and_then(|u| u.host_str().map(str::to_owned)).unwrap_or_else(|| url.to_owned())
Url::parse(url).ok().and_then(|u| u.host_str().map(str::to_owned)).unwrap_or_else(|| url.to_owned())
}
async fn rm(ctx: &Ctx, feed: &str) -> Result<()> {
@@ -1026,7 +1041,7 @@ async fn rm(ctx: &Ctx, feed: &str) -> Result<()> {
Ok(())
}
async fn import(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
async fn import(ctx: &Ctx, file: &Path) -> Result<()> {
let text = std::fs::read_to_string(file).with_context(|| format!("reading {}", file.display()))?;
// The CLI speaks for the operator, as the shared web token does.
let admin = ctx
@@ -1070,7 +1085,7 @@ pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i6
let mut cfg = ctx.edit_cfg().await;
// By address, once: each feed looked through every subscription and the whole catalogue, and
// the catalogue's lock is held throughout (#338).
let mut known = std::collections::HashMap::new();
let mut known = HashMap::new();
for s in subscriptions(ctx).await? {
// The first with an address wins, as a search in order found it.
known.entry(s.cfg.url).or_insert(s.id);
@@ -1131,7 +1146,7 @@ pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i6
/// database for the feeds and the server settings (issue #18). The first time a database holds
/// neither, it takes them from config.toml, which is then cut down to the rest, the original kept
/// beside it as config.toml.pre-database.
async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &std::path::Path) -> Result<config::Config> {
async fn assemble_config(db: &db::Db, mut cfg: config::Config, path: &Path) -> Result<config::Config> {
for _ in 0..2 {
if let Some((stored, feeds)) = db.stored_config().await? {
if config::Config::file_holds_stored(path) {
@@ -2142,9 +2157,8 @@ async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(String, String)]
let mut added = vec![];
// Read once, not once a new feed: each read is the whole feeds table.
let known = ctx.db.feed_urls().await?;
let fixed: std::collections::HashSet<&str> =
cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect();
let known_ids: std::collections::HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect();
let fixed: HashSet<&str> = cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect();
let known_ids: HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect();
for (title, url) in listed {
// Already known, whether derived or promoted into the config.
@@ -2325,7 +2339,7 @@ async fn fetch_one(
enclosure: i64,
feed_cfg: &config::Feed,
url: &str,
dest_dir: &std::path::Path,
dest_dir: &Path,
) -> Result<Option<(PathBuf, u64)>> {
// Throttled to whole percents, as the original's lastDLStepSize guard did.
let mut last_pct = -1i64;
@@ -2485,7 +2499,7 @@ async fn torrent_one(
feed_id: &str,
enclosure: i64,
url: &str,
dest_dir: &std::path::Path,
dest_dir: &Path,
) -> Result<(PathBuf, u64)> {
let name = download::filename_for(url, None);
let mut last_pct = -1i64;

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

@@ -6,10 +6,16 @@
//! Prometheus exporters have none: checking a token is a database lookup. Written out by hand:
//! the text format is a few lines, and a client library would be a dependency for them.
use axum::Router;
use std::collections::BTreeMap;
use std::fmt::Write as _;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::Ordering::Relaxed;
use std::sync::atomic::{AtomicI64, AtomicU64};
use std::time::Instant;
use tokio::runtime::Handle;
use tower_http::compression::CompressionLayer;
/// What the daemon counts as it works, for /metrics to read without asking anything (#195).
/// Global: the web layer, the worker, downloads, artwork and DeadAir each add to it, and a handle
@@ -51,11 +57,11 @@ pub struct Stats {
/// What each person's Signals, icons and key checks used, and whose key paid (#278): items
/// judged, Signals they fit, and TypeSafe's input and output tokens. Floats: a call shared by
/// several people on the server's key is split between them.
people: std::sync::Mutex<std::collections::BTreeMap<(String, &'static str), [f64; 4]>>,
people: std::sync::Mutex<BTreeMap<(String, &'static str), [f64; 4]>>,
/// The worker's job now, and since when.
job: std::sync::Mutex<Option<(&'static str, std::time::Instant)>>,
job: std::sync::Mutex<Option<(&'static str, Instant)>>,
/// Jobs finished, by kind: how many, and their seconds all told.
jobs: std::sync::Mutex<std::collections::BTreeMap<&'static str, (u64, f64)>>,
jobs: std::sync::Mutex<BTreeMap<&'static str, (u64, f64)>>,
}
impl Stats {
@@ -80,9 +86,9 @@ impl Stats {
jev_calls: [const { AtomicU64::new(0) }; 5],
jev_micros: AtomicU64::new(0),
jev_paced_micros: AtomicU64::new(0),
people: std::sync::Mutex::new(std::collections::BTreeMap::new()),
people: std::sync::Mutex::new(BTreeMap::new()),
job: std::sync::Mutex::new(None),
jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()),
jobs: std::sync::Mutex::new(BTreeMap::new()),
}
}
@@ -99,7 +105,7 @@ impl Stats {
/// Marks the worker as on a job of `kind` until what it returns is dropped, however the job
/// ends: finished, failed or stopped.
pub fn job(&'static self, kind: &'static str) -> Job {
*self.job.lock().expect("a panic elsewhere poisoned this lock") = Some((kind, std::time::Instant::now()));
*self.job.lock().expect("a panic elsewhere poisoned this lock") = Some((kind, Instant::now()));
Job(self)
}
}
@@ -143,12 +149,12 @@ pub struct Sources {
pub events: tokio::sync::broadcast::Sender<crate::ipc::Event>,
pub playing: crate::deadair::Playing,
/// The daemon's runtime, not the metrics thread's own.
pub runtime: tokio::runtime::Handle,
pub started: std::time::Instant,
pub runtime: Handle,
pub started: Instant,
}
/// Starts serving /metrics on `bind`, on a thread of its own, and says where.
pub fn spawn(bind: &str, s: Sources) -> anyhow::Result<std::net::SocketAddr> {
pub fn spawn(bind: &str, s: Sources) -> anyhow::Result<SocketAddr> {
use anyhow::Context as _;
// Bound here, so a port already taken is the daemon's error at start, not a thread's later.
let listener = std::net::TcpListener::bind(bind).with_context(|| format!("binding metrics to {bind}"))?;
@@ -161,7 +167,7 @@ pub fn spawn(bind: &str, s: Sources) -> anyhow::Result<std::net::SocketAddr> {
Err(e) => return tracing::error!(error = %e, "metrics could not start"),
};
rt.block_on(async move {
let app = axum::Router::new()
let app = Router::new()
.route(
"/metrics",
axum::routing::get(move || {
@@ -170,7 +176,7 @@ pub fn spawn(bind: &str, s: Sources) -> anyhow::Result<std::net::SocketAddr> {
}),
)
// Prometheus asks for gzip, every 15 s (#253).
.layer(tower_http::compression::CompressionLayer::new().gzip(true));
.layer(CompressionLayer::new().gzip(true));
match tokio::net::TcpListener::from_std(listener) {
Ok(l) => {
if let Err(e) = axum::serve(l, app).await {

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

@@ -6,6 +6,8 @@ use tracing::Instrument;
use crate::config::Config;
use crate::db;
use crate::db::{Candidate, Db};
use std::io::ErrorKind;
use std::path::Path;
/// Headroom kept below the quota, as getToQuota()'s `1048576 * 50` did.
pub const PAD_BYTES: u64 = 50 * 1024 * 1024;
@@ -103,8 +105,8 @@ async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> {
// A torrent's folder measures as a directory entry, not what is in it: the recorded size then.
let size = std::fs::metadata(&c.path).ok().filter(|m| m.is_file()).map_or(c.bytes.max(0) as u64, |m| m.len());
// Missing is fine: the point is that it is gone.
if let Err(e) = crate::download::remove(std::path::Path::new(&c.path))
&& e.kind() != std::io::ErrorKind::NotFound
if let Err(e) = crate::download::remove(Path::new(&c.path))
&& e.kind() != ErrorKind::NotFound
{
tracing::warn!(path = c.path, error = %e, "could not delete");
return Ok(0);

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

@@ -19,12 +19,16 @@ use crate::db;
use crate::db::{Ask, Unjudged};
use crate::feed;
use crate::metrics::STATS;
use std::collections::BTreeMap;
use std::fmt::{Display, Formatter};
use std::future::Future;
use std::time::Duration;
/// Items put to Jev a pass. A pass ends early once there are none left, so this only says how
/// long the queue is held before people who got a key later are looked at.
const BATCH: i64 = 50;
/// Between passes when nothing wakes it: a scan or a new Signal does, so this is a backstop.
const IDLE: std::time::Duration = std::time::Duration::from_secs(300);
const IDLE: Duration = Duration::from_secs(300);
/// Calls to Jev in flight at once (#275). One at a time, about 150 ms each, a new Signal's month
/// of items took twenty minutes. Enough for `MAX_RATE` at that pace; the pace is `pace`'s to keep.
const IN_FLIGHT: usize = 10;
@@ -103,7 +107,7 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
tracing::Span::current().record("calls", groups.len());
let refused = std::sync::Mutex::new(HashSet::new());
let stopped = AtomicBool::new(false);
let mut hits_for = std::collections::BTreeMap::<i64, usize>::new();
let mut hits_for = BTreeMap::<i64, usize>::new();
let mut judged = 0;
// Each call starts only when a place is free, so one that sees a key refused or the pass
// stopped is not sent; those already in flight are let finish, and their answers kept.
@@ -203,8 +207,8 @@ struct Refused {
body: String,
}
impl std::fmt::Display for Refused {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
impl Display for Refused {
fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
write!(f, "TypeSafe answered {}: {}", self.status, self.body)
}
}
@@ -238,7 +242,7 @@ fn slot(
tokens: f64,
) -> (tokio::time::Instant, tokio::time::Instant) {
let start = next.max(now);
(start, start + std::time::Duration::from_secs_f64((1.0 / MAX_RATE).max(tokens / MAX_TOKENS)))
(start, start + Duration::from_secs_f64((1.0 / MAX_RATE).max(tokens / MAX_TOKENS)))
}
/// Waits for this call's turn. Its tokens are guessed from the request, a token to three bytes
@@ -261,7 +265,7 @@ async fn pace(body: &serde_json::Value) {
/// TypeSafe said too many: nobody calls again until it says, or a second.
async fn back_off(retry_after: Option<u64>) {
let until = tokio::time::Instant::now() + std::time::Duration::from_secs(retry_after.unwrap_or(1).min(60));
let until = tokio::time::Instant::now() + Duration::from_secs(retry_after.unwrap_or(1).min(60));
let mut next = NEXT_CALL.lock().await;
*next = Some(next.map_or(until, |n| n.max(until)));
}
@@ -435,7 +439,7 @@ fn usage(answer: &serde_json::Value) -> [f64; 2] {
/// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span
/// failed, which a field named `error` alone does not: Tempo's failed traces left Jev's out.
async fn called<T>(span: tracing::Span, call: impl std::future::Future<Output = Result<T>>) -> Result<T> {
async fn called<T>(span: tracing::Span, call: impl Future<Output = Result<T>>) -> Result<T> {
let started = std::time::Instant::now();
let r = call.instrument(span.clone()).await;
STATS.jev_micros.fetch_add(started.elapsed().as_micros() as u64, Relaxed);

5
crates/ipx/src/torrent.rs
View File

@@ -9,6 +9,7 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use crate::config::Config;
use librqbit::{ListenerOptions, ManagedTorrent};
pub struct Torrents {
session: Arc<Session>,
@@ -26,7 +27,7 @@ impl Torrents {
let listen_addr: SocketAddr = format!("0.0.0.0:{lo}").parse()?;
let opts = SessionOptions {
listen: Some(librqbit::ListenerOptions { listen_addr, ..Default::default() }),
listen: Some(ListenerOptions { listen_addr, ..Default::default() }),
client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()),
..Default::default()
};
@@ -127,7 +128,7 @@ impl Torrents {
}
/// Seeds until the ratio or the time limit is reached, whichever comes first.
async fn seed(&self, cfg: &Config, handle: &Arc<librqbit::ManagedTorrent>) {
async fn seed(&self, cfg: &Config, handle: &Arc<ManagedTorrent>) {
if cfg.torrent.seed_ratio <= 0.0 || cfg.torrent.seed_time_mins == 0 {
return;
}

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

@@ -23,7 +23,16 @@ use tower_http::services::ServeFile;
use crate::Ctx;
use crate::db;
use crate::ipc::{Command, Event};
use axum::Extension;
use axum::extract::{ConnectInfo, FromRequestParts};
use axum::http::request::Parts;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::LazyLock;
use std::time::{Duration, Instant};
use tokio::net::TcpListener;
use tracing::Instrument;
use url::Url;
const COOKIE: &str = "ipx_token";
@@ -98,14 +107,13 @@ pub async fn serve(state: WebState, bind: &str) -> Result<()> {
use axum::serve::ListenerExt;
// Nagle off, as axum 0.7 had it: under load a response goes out in parts, and with Nagle
// on its last small part waits for the browser's delayed ACK, up to 40 ms (#170).
let listener =
tokio::net::TcpListener::bind(bind).await.with_context(|| format!("binding {bind}"))?.tap_io(|tcp| {
let listener = TcpListener::bind(bind).await.with_context(|| format!("binding {bind}"))?.tap_io(|tcp| {
if let Err(e) = tcp.set_nodelay(true) {
tracing::debug!(error = %e, "could not set TCP_NODELAY");
}
});
tracing::info!(bind, "web ui listening");
axum::serve(listener, router(state).into_make_service_with_connect_info::<std::net::SocketAddr>())
axum::serve(listener, router(state).into_make_service_with_connect_info::<SocketAddr>())
.await
.context("serving the web ui")
}
@@ -240,7 +248,7 @@ async fn vouched_name(
state: &WebState,
cfg: &crate::config::Config,
peer: String,
headers: &axum::http::HeaderMap,
headers: &HeaderMap,
) -> Option<String> {
if cfg.web.trusted_header.is_empty() || !cfg.web.trusted_proxies.iter().any(|p| p == &peer) {
return None;
@@ -257,21 +265,15 @@ async fn vouched_name(
}
fn peer(req: &Request) -> String {
req.extensions()
.get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
.map(|c| c.0.ip().to_string())
.unwrap_or_default()
req.extensions().get::<ConnectInfo<SocketAddr>>().map(|c| c.0.ip().to_string()).unwrap_or_default()
}
/// Handlers take `User` to say they need one; the auth layer put it there, and nothing
/// reaches a handler without passing through it.
impl<S: Send + Sync> axum::extract::FromRequestParts<S> for crate::db::User {
impl<S: Send + Sync> FromRequestParts<S> for crate::db::User {
type Rejection = (StatusCode, &'static str);
async fn from_request_parts(
parts: &mut axum::http::request::Parts,
_state: &S,
) -> std::result::Result<Self, Self::Rejection> {
async fn from_request_parts(parts: &mut Parts, _state: &S) -> std::result::Result<Self, Self::Rejection> {
parts.extensions.get::<crate::db::User>().cloned().ok_or((StatusCode::UNAUTHORIZED, "sign in"))
}
}
@@ -346,7 +348,7 @@ async fn logout(State(state): State<WebState>, req: Request) -> Response {
async fn me(
State(state): State<WebState>,
user: crate::db::User,
axum::Extension(Proxied(by_proxy)): axum::Extension<Proxied>,
Extension(Proxied(by_proxy)): Extension<Proxied>,
) -> Json<serde_json::Value> {
let url = state.ctx.cfg().web.sign_out_url.clone();
let sign_out = (by_proxy && !url.is_empty()).then_some(url);
@@ -502,17 +504,12 @@ struct IconAsk {
/// the route is not tied to a Signal, so a loop on it ran up the server's bill and held back
/// everyone's Signals behind the shared pace (#340). A Signal is named a few times at most.
const ICON_PICKS_AN_HOUR: u32 = 20;
static ICON_PICKS: std::sync::LazyLock<std::sync::Mutex<std::collections::HashMap<i64, (std::time::Instant, u32)>>> =
std::sync::LazyLock::new(Default::default);
static ICON_PICKS: LazyLock<std::sync::Mutex<HashMap<i64, (Instant, u32)>>> = LazyLock::new(Default::default);
/// Counts a pick against the person's hour, which starts at their first pick in it.
fn icon_pick_allowed(
picks: &mut std::collections::HashMap<i64, (std::time::Instant, u32)>,
user: i64,
now: std::time::Instant,
) -> bool {
fn icon_pick_allowed(picks: &mut HashMap<i64, (Instant, u32)>, user: i64, now: Instant) -> bool {
let (start, used) = picks.entry(user).or_insert((now, 0));
if now.duration_since(*start) >= std::time::Duration::from_secs(3600) {
if now.duration_since(*start) >= Duration::from_secs(3600) {
(*start, *used) = (now, 0);
}
if *used >= ICON_PICKS_AN_HOUR {
@@ -535,7 +532,7 @@ async fn pick_signal_icon(
if !icon_pick_allowed(
&mut ICON_PICKS.lock().expect("a panic elsewhere poisoned this lock"),
user.id,
std::time::Instant::now(),
Instant::now(),
) {
return Err(ApiError::too_many("enough icons picked for one hour; choose one by hand"));
}
@@ -899,7 +896,7 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>)
// Config entries plus the feeds derived from OPML subscriptions -- the catalogue.
// What comes back is only the part of it this person subscribes to.
let subs = crate::subscriptions(&state.ctx).instrument(db::step("catalogue")).await?;
let mine: std::collections::HashMap<String, crate::db::Sub> = state
let mine: HashMap<String, crate::db::Sub> = state
.ctx
.db
.subscriptions_for(user.id)
@@ -913,9 +910,8 @@ async fn feed_rows(state: &WebState, user: &crate::db::User, only: Option<&str>)
let kept = state.ctx.db.kept_feeds(user.id).instrument(db::step("kept")).await?;
let (hidden, quiet) = state.ctx.db.hidden_from(user.id).instrument(db::step("hidden_from")).await?;
let prefs = state.ctx.db.deadair_prefs(Some(user.id)).await?;
let deadair: std::collections::HashMap<&str, bool> = prefs.iter().map(|(_, f, w, _)| (f.as_str(), *w)).collect();
let pauses: std::collections::HashMap<&str, i64> =
prefs.iter().filter_map(|(_, f, _, p)| Some((f.as_str(), (*p)?))).collect();
let deadair: HashMap<&str, bool> = prefs.iter().map(|(_, f, w, _)| (f.as_str(), *w)).collect();
let pauses: HashMap<&str, i64> = prefs.iter().filter_map(|(_, f, _, p)| Some((f.as_str(), (*p)?))).collect();
let mut listed = state.ctx.db.feed_list(user.id, only).instrument(db::step("feed_list")).await?;
let mut out = Vec::with_capacity(mine.len());
let asked = subs.iter().filter(|s| only.is_none_or(|o| o == s.id));
@@ -1034,7 +1030,7 @@ fn dir_error(e: crate::directory::DirError) -> ApiError {
/// address, since the directory and this server name a feed differently.
async fn rows_for(state: &WebState, user_id: i64, listed: Vec<ipx_directory::Listing>) -> Result<Vec<PopularRow>> {
// A step each, as the feed list has: a Directory page of 983 ms showed 19 under spans (#250).
let mine: std::collections::HashSet<String> = state
let mine: HashSet<String> = state
.ctx
.db
.subscriptions_for(user_id)
@@ -1043,7 +1039,7 @@ async fn rows_for(state: &WebState, user_id: i64, listed: Vec<ipx_directory::Lis
.into_iter()
.map(|s| s.feed_id)
.collect();
let here: std::collections::HashMap<String, String> = crate::subscriptions(&state.ctx)
let here: HashMap<String, String> = crate::subscriptions(&state.ctx)
.instrument(db::step("catalogue"))
.await?
.into_iter()
@@ -1232,16 +1228,12 @@ async fn subscribe_popular(
/// Validates a replacement feed URL: present, parseable, and not already subscribed under
/// a different id. Returns the trimmed URL.
fn check_url(
url: &str,
id: &str,
feeds: &std::collections::BTreeMap<String, crate::config::Feed>,
) -> Result<String, String> {
fn check_url(url: &str, id: &str, feeds: &BTreeMap<String, crate::config::Feed>) -> Result<String, String> {
let url = url.trim();
if url.is_empty() {
return Err("the feed URL cannot be empty".into());
}
match url::Url::parse(url) {
match Url::parse(url) {
Ok(u) if u.scheme() == "http" || u.scheme() == "https" => {}
Ok(u) => return Err(format!("{:?} is not an http(s) URL", u.scheme())),
Err(e) => return Err(format!("that is not a valid URL: {e}")),