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

Faster scans and directory rounds, and fetches named by feed (#227, #228, #229, #230)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-14 13:27:19 +00:00
parent cece114d98
commit 8d7a30d1f9
7 changed files with 72 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -1862,6 +1862,7 @@ dependencies = [
"anyhow",
"axum",
"clap",
"futures-util",
"ipx-feed",
"mimalloc",
"opentelemetry",

5
crates/directory/CHANGELOG.md
View File

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

1
crates/directory/Cargo.toml
View File

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

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

@@ -134,6 +134,8 @@ impl Directory {
pub fn new(store: Store, tokens: Vec<Token>) -> 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))
.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");
},

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

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

6
crates/ipx/CHANGELOG.md
View File

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

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

@@ -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<Ctx>, 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<Result<(feed::Fetched, Option<String>)>>,
// 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<String>)>, Option<String>)>,
) -> Result<Outcome> {
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"<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title><link>http://127.0.0.1:9/</link><item><title>One</title><guid>g1</guid></item></channel></rss>".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;