//! 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, etag: Option, last_modified: Option }, } /// 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. Without a limit an episode's MP3 given as a feed was read /// whole into memory, at every check (#295). Real feeds can be large: a daily show's whole back /// catalogue was 62 MB, and 50 MB here stopped it updating (#357). pub(crate) const MAX_BODY: usize = 200 << 20; /// How many times a YouTube feed's spurious failure is asked again, each after its turn. const YOUTUBE_RETRIES: u32 = 2; /// The least time between two requests for YouTube feeds from this process. A scan asked for six /// at once, and channels in a row failed together while one asked alone a minute later read /// (#386); asked again a second later, they failed again. const YOUTUBE_GAP: Duration = Duration::from_secs(2); static YOUTUBE_NEXT: tokio::sync::Mutex> = tokio::sync::Mutex::const_new(None); /// Waits until `gap` after the last request's turn, one caller at a time: tokio's lock is fair, /// so waiting feeds go in the order they asked. shortcut: 39 channels due together add up to 80 s /// to that scan; a scan's other feeds wait behind them only when they come after in the order. #[tracing::instrument(name = "youtube_turn", skip_all)] async fn take_turn(next: &tokio::sync::Mutex>, gap: Duration) { let mut next = next.lock().await; if let Some(at) = *next { tokio::time::sleep_until(at).await; } *next = Some(tokio::time::Instant::now() + gap); } /// YouTube answers some feed requests with a 404 or 500 it does not mean, in tens of /// milliseconds, while the next request for the same feed reads; checked once a day, a channel /// then missed most days (#382). fn youtube_spurious(url: &reqwest::Url, status: StatusCode) -> bool { is_youtube_feed(url.as_str()) && (status == StatusCode::NOT_FOUND || status.is_server_error()) } /// A YouTube channel's or playlist's feed, which fails most reads in streaks that outlast /// `fetch`'s retries, so a scheduler asks it again sooner than other feeds (#386). #[must_use] pub fn is_youtube_feed(url: &str) -> bool { reqwest::Url::parse(url).is_ok_and(|u| u.host_str() == Some("www.youtube.com") && u.path() == "/feeds/videos.xml") } /// 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> { 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)>, etag: Option<&str>, last_modified: Option<&str>, ) -> Result<(Fetched, Option)> { let start = reqwest::Url::parse(feed_url).context("the feed's address")?; let mut url = start.clone(); let mut permanent = true; let mut retries = YOUTUBE_RETRIES; // Ten redirects, and the retries beside them. for _ in 0..10 + YOUTUBE_RETRIES { if is_youtube_feed(url.as_str()) { take_turn(&YOUTUBE_NEXT, YOUTUBE_GAP).await; } 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 retries > 0 && youtube_spurious(&url, status) { retries -= 1; continue; } 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 = "T"; 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 youtube_turns_are_a_gap_apart_one_at_a_time() { let (next, gap) = (tokio::sync::Mutex::const_new(None), std::time::Duration::from_millis(50)); let start = tokio::time::Instant::now(); // The first goes at once; three asking together then go one gap after another. super::take_turn(&next, gap).await; assert!(start.elapsed() < gap); let t = || async { super::take_turn(&next, gap).await; start.elapsed() }; let (a, b, c) = tokio::join!(t(), t(), t()); let mut at = [a, b, c]; at.sort(); for (i, at) in at.iter().enumerate() { assert!(*at >= gap * (i as u32 + 1), "turn {} at {at:?}", i + 1); } } #[test] fn only_a_youtube_feeds_404_or_5xx_is_asked_again() { use reqwest::StatusCode; let yt = reqwest::Url::parse("https://www.youtube.com/feeds/videos.xml?user=x").unwrap(); let other = reqwest::Url::parse("https://example.com/feeds/videos.xml").unwrap(); let page = reqwest::Url::parse("https://www.youtube.com/watch?v=x").unwrap(); assert!(super::youtube_spurious(&yt, StatusCode::NOT_FOUND)); assert!(super::youtube_spurious(&yt, StatusCode::INTERNAL_SERVER_ERROR)); assert!(!super::youtube_spurious(&yt, StatusCode::FORBIDDEN)); assert!(!super::youtube_spurious(&other, StatusCode::NOT_FOUND)); assert!(!super::youtube_spurious(&page, StatusCode::NOT_FOUND)); } #[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"); } }