From 8d7a30d1f95ea99bd74206d8c156a8b9bdd58e32 Mon Sep 17 00:00:00 2001 From: rays Date: Wed, 14 Oct 2026 13:27:19 +0000 Subject: [PATCH] Faster scans and directory rounds, and fetches named by feed (#227, #228, #229, #230) From the slow traces of 2026-10-13: the directory checked a round's 25 feeds one after another, so two libsyn feeds timing out at 30 s made a round 70 s; a forced scan looked up 62 site icons in turn, 22 s of 62; mjg59.dreamwidth.org takes the connection and never answers, so it ran the full 30 s in every scan it was in; and since #225 strips queries, every YouTube fetch span read the same address, outside any span naming its feed. The directory now checks five at a time (SQLite stays on one connection, so its writes queue). A forced scan looks up the icon in the fetch-ahead task, which already runs six deep, and hands it to scan_one with the body. Both feed clients give up after 15 s of silence (no successful fetch took over 8 s in a day of traces); FEED_TIMEOUT stays 30 s for a feed still sending. The directory's client had no connect timeout; it has ipx's 10 s. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/directory/CHANGELOG.md | 5 ++++ crates/directory/Cargo.toml | 1 + crates/directory/src/lib.rs | 15 ++++++++--- crates/feed/src/lib.rs | 6 +++++ crates/ipx/CHANGELOG.md | 6 +++++ crates/ipx/src/main.rs | 48 ++++++++++++++++++++++++++++++----- 7 files changed, 72 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 60c1b5a..3b0dd26 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1862,6 +1862,7 @@ dependencies = [ "anyhow", "axum", "clap", + "futures-util", "ipx-feed", "mimalloc", "opentelemetry", diff --git a/crates/directory/CHANGELOG.md b/crates/directory/CHANGELOG.md index cb628e1..03263c5 100644 --- a/crates/directory/CHANGELOG.md +++ b/crates/directory/CHANGELOG.md @@ -10,6 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- The directory checks five feeds at a time, so a publisher that does not answer no longer holds up the rest of its round (#227). +- A feed's server that takes 10 s to connect, or sends nothing for 15 s, is given up on (#230). + ## [1.1.0] - 2026-10-13 ### Added diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index e375235..45ba019 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -12,6 +12,7 @@ ammonia.workspace = true anyhow.workspace = true axum.workspace = true clap.workspace = true +futures-util.workspace = true ipx-feed.workspace = true opentelemetry.workspace = true opentelemetry-otlp.workspace = true diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 74a6e06..e0d3d13 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -134,6 +134,8 @@ impl Directory { pub fn new(store: Store, tokens: Vec) -> Result { // No redirects: ipx_feed::fetch follows them itself, so it sees each hop. let client = reqwest::Client::builder() + .connect_timeout(std::time::Duration::from_secs(10)) + .read_timeout(ipx_feed::FEED_SILENCE) .redirect(reqwest::redirect::Policy::none()) .user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION"))) .build()?; @@ -193,12 +195,17 @@ pub async fn refresher(dir: Shared) { let round = tracing::info_span!("refresh", feeds = due.len()); tracing::Instrument::instrument( async { - for d in &due { + // 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); - if let Err(e) = tracing::Instrument::instrument(dir.refresh(d), check).await { - tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check"); + async { + if let Err(e) = tracing::Instrument::instrument(dir.refresh(d), check).await { + tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check"); + } } - } + }) + .await; // One line a round, for the dashboard's refresher panel (#210); failures have their own. tracing::info!(checked = due.len(), "feeds checked"); }, diff --git a/crates/feed/src/lib.rs b/crates/feed/src/lib.rs index 9d984f3..6b638c1 100644 --- a/crates/feed/src/lib.rs +++ b/crates/feed/src/lib.rs @@ -61,6 +61,12 @@ pub enum Fetched { /// downloads, which are not, have no such limit. pub const FEED_TIMEOUT: std::time::Duration = std::time::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); + /// 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. diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 9e1b568..f628783 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -12,6 +12,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- Check every feed looks up the site icons of feeds without artwork a few at a time, alongside the fetches, instead of one after another (#228). +- A feed's server that sends nothing for 15 s is given up on, instead of holding a scan for the full 30 s; a feed that keeps sending still has 30 s (#230). +- A scan's trace names the feed each fetch is for (#229). + ## [1.1.0] - 2026-10-13 ### Added diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index b4603b6..6d97562 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -355,6 +355,7 @@ async fn main() -> Result<()> { feed_client: reqwest::Client::builder() .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) .connect_timeout(std::time::Duration::from_secs(10)) + .read_timeout(feed::FEED_SILENCE) .redirect(reqwest::redirect::Policy::none()) .build()?, out: if is_daemon { Emitter::socket(events.clone(), false) } else { Emitter::terminal() }, @@ -1260,18 +1261,33 @@ async fn fetch(ctx: &Arc, only: Option<&str>, force: bool, scope: &[String] const AHEAD: usize = 6; let mut bodies = futures_util::StreamExt::buffered( futures_util::stream::iter(due.iter().map(|(sub, state)| { - let (client, feed_cfg) = (ctx.feed_client.clone(), sub.cfg.clone()); + let (client, web, feed_cfg) = (ctx.feed_client.clone(), ctx.client.clone(), sub.cfg.clone()); let (etag, modified) = (state.etag.clone(), state.last_modified.clone()); let early = !feed::is_patreon_creator(&feed_cfg.url); + // Named for its feed: the fetch's own span has only the address, without its query + // since #225, which for every YouTube channel is the same (#229). + let span = tracing::info_span!("fetch_ahead", feed = %sub.id); tokio::spawn(tracing::Instrument::instrument( async move { if !early { return None; } let _fetching = metrics::busy(&metrics::STATS.feeds_fetching); - Some(feed::fetch(&client, &feed_cfg.url, feed_cfg.auth(), etag.as_deref(), modified.as_deref()).await) + let got = feed::fetch(&client, &feed_cfg.url, feed_cfg.auth(), etag.as_deref(), modified.as_deref()).await; + // A forced scan looks for the site's icon of each feed naming no artwork: here, + // a few feeds at a time, not each in its turn, 22 s of a 62 s scan (#228). + let icon = match &got { + Ok((feed::Fetched::Body { bytes, .. }, _)) if force => match feed::parse(bytes) { + Ok(feed::ParsedFeed { image: None, site: Some(site), .. }) => { + Some(feed::site_icon(&web, &site).await.unwrap_or_default()) + } + _ => None, + }, + _ => None, + }; + Some((got, icon)) }, - tracing::Span::current(), + span, )) })), AHEAD, @@ -1628,9 +1644,14 @@ async fn scan_one( feed_cfg: &config::Feed, state: &db::HttpState, force: bool, - // The body the scan already fetched for it, if it did (#104), and where it moved, if it did. - prefetched: Option)>>, + // The body the scan already fetched for it, if it did (#104), and where it moved, if it did; + // with the site's icon when a forced scan looked it up meanwhile, "" for none (#228). + prefetched: Option<(Result<(feed::Fetched, Option)>, Option)>, ) -> Result { + let (prefetched, icon) = match prefetched { + Some((got, icon)) => (Some(got), icon), + None => (None, None), + }; // The queue follows the feed's settings each time it is due, changed or not, so files a // limit or auto-download no longer reaches stop counting as waiting (#113). let policy = policy_for(ctx, id, feed_cfg).await?; @@ -1738,7 +1759,10 @@ async fn scan_one( if parsed.image.is_none() { parsed.image = Some(match (&stored.image, &parsed.site) { (Some(known), _) if !force => known.clone(), - (_, Some(site)) => feed::site_icon(&ctx.client, site).await.unwrap_or_default(), + (_, Some(site)) => match &icon { + Some(found) => found.clone(), + None => feed::site_icon(&ctx.client, site).await.unwrap_or_default(), + }, (_, None) => String::new(), }); } @@ -2758,6 +2782,18 @@ mod tests { assert_eq!(text().await, ["As it should read."]); } + #[tokio::test] + async fn a_forced_scan_takes_the_icon_looked_up_ahead() { + let ctx = Arc::new(test_ctx(config::Config::default()).await); + // No artwork, and a site nothing answers at: only the icon found ahead can be stored (#228). + let body = b"Thttp://127.0.0.1:9/Oneg1".to_vec(); + let got = feed::Fetched::Body { bytes: body, etag: None, last_modified: None }; + let icon = "https://example.com/favicon.ico".to_string(); + let cfg = config::Feed { url: "http://127.0.0.1:9/feed".into(), ..feed() }; + scan_one(&ctx, "iconic", &cfg, &db::HttpState::default(), true, Some((Ok((got, None)), Some(icon.clone())))).await.unwrap(); + assert_eq!(ctx.db.feed_summary("iconic").await.unwrap().image, Some(icon)); + } + #[tokio::test] async fn nothing_from_a_feed_that_never_posted_is_an_error_and_from_one_that_has_is_not() { let base = empty_or_feed_server().await;