ipx-feed's lib.rs splits into leaf modules (#356)
lib.rs held 2,000 lines: fetching, parsing, OPML, Patreon, discovery, artwork, categories, Jev and redaction, and their tests. It is now the module list and the re-exports, so ipx and the directory call feed::fetch, feed::parse and the rest as before: fetch.rs fetch and its limits parse.rs RSS and Atom outlines.rs OPML (not opml.rs, which would shadow the opml crate) patreon.rs a creator's shows discover.rs finding a feed wordpress.rs WordPress.com's API failure.rs failures in words privacy.rs redact, looks_private artwork.rs https art, icons category.rs Apple's categories jev.rs Jev's category Moved verbatim, private items pub(crate) where a sibling calls them, calls across modules module::function. Every definition and doc comment line of the old lib.rs is in exactly one of the new files; each test went with what it tests, and all 40 run. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
189
crates/feed/src/fetch.rs
Normal file
189
crates/feed/src/fetch.rs
Normal file
@@ -0,0 +1,189 @@
|
||||
//! Reading a feed over HTTP: conditional requests, redirects followed one by one, bounded bodies.
|
||||
|
||||
use crate::{privacy, wordpress};
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
use reqwest::StatusCode;
|
||||
use reqwest::header::HeaderName;
|
||||
use reqwest::header::{ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED};
|
||||
use std::sync::atomic::AtomicU64;
|
||||
use std::time::Duration;
|
||||
|
||||
/// What `fetch` got.
|
||||
pub enum Fetched {
|
||||
/// Server said 304, or returned a body we already have.
|
||||
NotModified,
|
||||
/// A body to read, with the validators to send next time.
|
||||
Body { bytes: Vec<u8>, etag: Option<String>, last_modified: Option<String> },
|
||||
}
|
||||
|
||||
/// The longest a feed may take, connecting to the last byte: a scan handles feeds in order, and
|
||||
/// 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: 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: 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: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
/// See `FETCHES`.
|
||||
pub static FETCHED_BYTES: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
/// The most of a feed or page read. The largest feeds here are a few MB; without a limit an
|
||||
/// episode's MP3 given as a feed was read whole into memory, at every check (#295).
|
||||
pub(crate) const MAX_BODY: usize = 50 << 20;
|
||||
|
||||
/// A response's body, up to `MAX_BODY`: past it, an error, and the rest is never read.
|
||||
pub(crate) async fn read_capped(mut resp: reqwest::Response) -> Result<Vec<u8>> {
|
||||
if resp.content_length().is_some_and(|n| n > MAX_BODY as u64) {
|
||||
return Err(anyhow!("larger than {} MB, too large to be a feed", MAX_BODY >> 20));
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
while let Some(c) = resp.chunk().await? {
|
||||
out.extend_from_slice(&c);
|
||||
anyhow::ensure!(out.len() <= MAX_BODY, "larger than {} MB, too large to be a feed", MAX_BODY >> 20);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Reads a feed, asking only for what changed since `etag` and `last_modified`. Redirects are
|
||||
/// followed here, up to ten, so `auth` goes to the feed's own host alone. Also returns the new
|
||||
/// address when every hop said the feed moved for good, for the caller to keep instead.
|
||||
#[tracing::instrument(skip_all, fields(url = %privacy::redact(feed_url), http.response.body.size))]
|
||||
pub async fn fetch(
|
||||
client: &reqwest::Client,
|
||||
feed_url: &str,
|
||||
auth: Option<(&str, Option<String>)>,
|
||||
etag: Option<&str>,
|
||||
last_modified: Option<&str>,
|
||||
) -> Result<(Fetched, Option<String>)> {
|
||||
let start = reqwest::Url::parse(feed_url).context("the feed's address")?;
|
||||
let mut url = start.clone();
|
||||
let mut permanent = true;
|
||||
for _ in 0..10 {
|
||||
let mut req = client.get(url.clone()).timeout(FEED_TIMEOUT);
|
||||
if let Some(tag) = etag {
|
||||
req = req.header(IF_NONE_MATCH, tag);
|
||||
}
|
||||
if let Some(lm) = last_modified {
|
||||
req = req.header(IF_MODIFIED_SINCE, lm);
|
||||
}
|
||||
// The feed's own host only: a redirect elsewhere must not be handed the password.
|
||||
if let Some((user, password)) = &auth
|
||||
&& url.host_str() == start.host_str()
|
||||
{
|
||||
req = req.basic_auth(user, password.as_deref());
|
||||
}
|
||||
|
||||
let resp = req.send().await.context("connecting")?;
|
||||
let status = resp.status();
|
||||
if status.is_redirection() && status != StatusCode::NOT_MODIFIED {
|
||||
let to = resp
|
||||
.headers()
|
||||
.get(reqwest::header::LOCATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.ok_or_else(|| anyhow!("HTTP {status} without a Location to go to"))?;
|
||||
url = url.join(to).with_context(|| format!("redirected to {to:?}, which is not an address"))?;
|
||||
permanent &= matches!(status, StatusCode::MOVED_PERMANENTLY | StatusCode::PERMANENT_REDIRECT);
|
||||
continue;
|
||||
}
|
||||
let moved = (permanent && url != start).then(|| url.to_string());
|
||||
FETCHES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
if status == StatusCode::NOT_MODIFIED {
|
||||
return Ok((Fetched::NotModified, moved));
|
||||
}
|
||||
if !status.is_success() {
|
||||
// WordPress.com answers some readers' feed requests with a browser check that only
|
||||
// JavaScript passes, for hours at a time and whatever the User-Agent; its API has the
|
||||
// same posts and no check (#317).
|
||||
if status == StatusCode::FORBIDDEN
|
||||
&& let Some(host) = url.host_str().map(str::to_owned)
|
||||
&& read_capped(resp).await.is_ok_and(|b| wordpress::is_wpcom_challenge(&b))
|
||||
{
|
||||
return Ok((wordpress::wpcom_posts(client, &host).await?, moved));
|
||||
}
|
||||
// The original surfaced 401/407 specially; the code is enough for a UI to switch on.
|
||||
return Err(anyhow!("HTTP {status}"));
|
||||
}
|
||||
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")?;
|
||||
tracing::Span::current().record("http.response.body.size", bytes.len());
|
||||
FETCHED_BYTES.fetch_add(bytes.len() as u64, std::sync::atomic::Ordering::Relaxed);
|
||||
return Ok((Fetched::Body { bytes, etag, last_modified }, moved));
|
||||
}
|
||||
// Worded as reqwest worded it, which failure_kind reads as a redirect loop.
|
||||
Err(anyhow!("error following redirect for url ({url}): too many redirects"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
|
||||
use crate::failure;
|
||||
|
||||
/// A server answering by path: /old moves for good to /new, /tmp for now, /chain for good
|
||||
/// to /tmp, /new is the feed.
|
||||
async fn redirecting_server() -> String {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = listener.accept().await else { return };
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 2048];
|
||||
let n = sock.read(&mut buf).await.unwrap_or(0);
|
||||
let req = String::from_utf8_lossy(&buf[..n]);
|
||||
let path = req.split_whitespace().nth(1).unwrap_or("/").to_owned();
|
||||
let body = "<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title></channel></rss>";
|
||||
let resp = match path.as_str() {
|
||||
"/old" => {
|
||||
"HTTP/1.1 301 Moved Permanently\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n".to_owned()
|
||||
}
|
||||
"/tmp" => "HTTP/1.1 302 Found\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n".to_owned(),
|
||||
"/chain" => {
|
||||
"HTTP/1.1 308 Permanent Redirect\r\nLocation: /tmp\r\nContent-Length: 0\r\n\r\n".to_owned()
|
||||
}
|
||||
// An episode's file given as a feed: its size said, its body never sent.
|
||||
"/huge" => "HTTP/1.1 200 OK\r\nContent-Length: 600000000\r\n\r\n".to_owned(),
|
||||
"/loop" => {
|
||||
"HTTP/1.1 301 Moved Permanently\r\nLocation: /loop\r\nContent-Length: 0\r\n\r\n".to_owned()
|
||||
}
|
||||
_ => format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{body}", body.len()),
|
||||
};
|
||||
let _ = sock.write_all(resp.as_bytes()).await;
|
||||
});
|
||||
}
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_feed_that_moved_for_good_says_where_and_one_moved_for_now_does_not() {
|
||||
let base = redirecting_server().await;
|
||||
let client = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build().unwrap();
|
||||
let get = |path: &str| {
|
||||
let url = format!("{base}{path}");
|
||||
let client = client.clone();
|
||||
async move { super::fetch(&client, &url, None, None, None).await }
|
||||
};
|
||||
let (got, moved) = get("/old").await.unwrap();
|
||||
assert!(matches!(got, super::Fetched::Body { .. }));
|
||||
assert_eq!(moved, Some(format!("{base}/new")));
|
||||
assert_eq!(get("/tmp").await.unwrap().1, None); // 302: for now
|
||||
let huge = get("/huge").await.err().map(|e| format!("{e:#}")).unwrap_or_default();
|
||||
assert!(huge.contains("too large to be a feed"), "{huge}");
|
||||
assert_eq!(get("/chain").await.unwrap().1, None); // 308 then 302: not for good
|
||||
assert_eq!(get("/new").await.unwrap().1, None); // never moved
|
||||
let looped = get("/loop").await.err().unwrap().to_string();
|
||||
assert_eq!(failure::failure_kind(&looped).0, "redirect_loop");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user