From 4783731c2eba80377793da9df7a3b70834cd7186 Mon Sep 17 00:00:00 2001 From: poach3r Date: Mon, 7 Sep 2026 15:04:41 -0400 Subject: [PATCH] replace round robin dispatch with a per-feed one --- src/cache.rs | 77 +++++++++++++++++++++++++++++ src/config.rs | 48 +++++++++++++++++- src/http.rs | 1 + src/http_fetch.rs | 122 +++++++++++++++++++++++++++++++++++++++++++++- src/opml.rs | 1 + src/rss.rs | 1 + src/scheduler.rs | 102 ++++++++++++++++++++++++++++++++------ 7 files changed, 332 insertions(+), 20 deletions(-) diff --git a/src/cache.rs b/src/cache.rs index b589480..c6c2068 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -11,6 +11,19 @@ fn timestamp_path(state_dir: &Path, name: &str) -> PathBuf { state_dir.join(format!("{name}.timestamp")) } +fn validators_path(state_dir: &Path, name: &str) -> PathBuf { + state_dir.join(format!("{name}.validators")) +} + +/// The conditional-request validators an origin server sent with a site's +/// last successful fetch, so the next fetch can ask "has this changed?" +/// instead of downloading the page again regardless. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct Validators { + pub etag: Option, + pub last_modified: Option, +} + /// Writes `path` by writing to a `.tmp` sibling and renaming over the /// destination, so a reader never observes a partially-written file. fn atomic_write(path: &Path, contents: &[u8]) -> io::Result<()> { @@ -42,3 +55,67 @@ pub fn read_timestamp(state_dir: &Path, name: &str) -> Option { .parse() .ok() } + +/// Persists the validators from a site's latest response so the next fetch +/// can send them back as `If-None-Match`/`If-Modified-Since`. Called +/// unconditionally after every successful (non-304) fetch, including when a +/// server stops sending a validator it previously sent -- the stored value +/// should track what the server currently supports. +pub fn write_validators(state_dir: &Path, name: &str, validators: &Validators) -> io::Result<()> { + fs::create_dir_all(state_dir)?; + let contents = format!( + "{}\n{}\n", + validators.etag.as_deref().unwrap_or(""), + validators.last_modified.as_deref().unwrap_or(""), + ); + atomic_write(&validators_path(state_dir, name), contents.as_bytes()) +} + +pub fn read_validators(state_dir: &Path, name: &str) -> Validators { + let contents = fs::read_to_string(validators_path(state_dir, name)).unwrap_or_default(); + let mut lines = contents.lines(); + let non_empty = |s: &str| (!s.is_empty()).then(|| s.to_string()); + Validators { + etag: lines.next().and_then(non_empty), + last_modified: lines.next().and_then(non_empty), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scratch_dir(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("luafeed-cache-test-{name}")); + let _ = fs::remove_dir_all(&dir); + dir + } + + #[test] + fn missing_validators_file_reads_as_all_none() { + let dir = scratch_dir("missing-validators"); + assert_eq!(read_validators(&dir, "site"), Validators::default()); + } + + #[test] + fn validators_round_trip_through_write_and_read() { + let dir = scratch_dir("validators-round-trip"); + let validators = Validators { + etag: Some("\"abc123\"".to_string()), + last_modified: Some("Sun, 06 Sep 2026 00:00:00 GMT".to_string()), + }; + write_validators(&dir, "site", &validators).unwrap(); + assert_eq!(read_validators(&dir, "site"), validators); + } + + #[test] + fn validators_round_trip_with_only_one_field_present() { + let dir = scratch_dir("validators-partial"); + let validators = Validators { + etag: Some("\"abc123\"".to_string()), + last_modified: None, + }; + write_validators(&dir, "site", &validators).unwrap(); + assert_eq!(read_validators(&dir, "site"), validators); + } +} diff --git a/src/config.rs b/src/config.rs index 769779b..da27565 100644 --- a/src/config.rs +++ b/src/config.rs @@ -8,6 +8,8 @@ use std::time::Duration; #[derive(Debug)] pub struct Config { + /// Default refetch interval for sites that don't set their own + /// `interval`. Also the fallback used when `interval` is absent. pub delay: Duration, pub listen_addr: String, pub listen_port: u16, @@ -24,6 +26,9 @@ pub struct SiteConfig { pub description: String, pub method_source: Arc, pub method_label: String, + /// How often this site should be refetched. Defaults to `Config::delay` + /// when the site doesn't set its own `interval`. + pub interval: Duration, } #[derive(Debug)] @@ -58,6 +63,10 @@ pub fn load_config(path: &Path) -> Result { pub fn load_config_from_str(source: &str, base_dir: &Path) -> Result { let lua = Lua::new(); let sites = RefCell::new(Vec::new()); + // Per-site `interval` overrides, parsed alongside `sites` (same index) + // but resolved against the global `delay` only after the whole script + // has run, since `delay` may be assigned after a `site()` call. + let interval_overrides: RefCell>> = RefCell::new(Vec::new()); lua.scope(|scope| { let site_fn = scope.create_function(|_, table: Table| { @@ -71,6 +80,7 @@ pub fn load_config_from_str(source: &str, base_dir: &Path) -> Result>("description")? .unwrap_or_default(); + let interval_secs: Option = table.get("interval")?; let method_source = method::resolve_method(&method_label, base_dir).map_err(|e| { mlua::Error::RuntimeError(format!( @@ -86,7 +96,9 @@ pub fn load_config_from_str(source: &str, base_dir: &Path) -> Result Result>("delay") .map_err(ConfigError::Lua)? - .unwrap_or(300); + .unwrap_or(1800); + + let mut sites = sites.into_inner(); + for (site, interval_secs) in sites.iter_mut().zip(interval_overrides.into_inner()) { + site.interval = Duration::from_secs(interval_secs.unwrap_or(delay_secs)); + } let listen_addr: String = globals .get::>("listen_addr") .map_err(ConfigError::Lua)? @@ -117,7 +134,7 @@ pub fn load_config_from_str(source: &str, base_dir: &Path) -> Result = LazyLock::new(|| { @@ -8,8 +11,51 @@ static CLIENT: LazyLock = LazyLock::new(|| { .expect("building the HTTP client should not fail") }); -pub fn fetch(url: &str) -> Result { - CLIENT.get(url).send()?.error_for_status()?.text() +pub enum FetchResult { + /// The server sent a fresh body, along with whatever validators it + /// attached for next time (either may be absent, or unchanged from what + /// was sent). + Modified { + body: String, + validators: Validators, + }, + /// The server confirmed (via 304) that `validators` still describes the + /// page: the caller's cached copy is still current and there's nothing + /// new to parse. + NotModified, +} + +/// Fetches `url`, sending `validators` as `If-None-Match`/`If-Modified-Since` +/// so a server that supports conditional requests can answer with a cheap +/// 304 instead of resending a page that hasn't changed. +pub fn fetch(url: &str, validators: &Validators) -> Result { + let mut request = CLIENT.get(url); + if let Some(etag) = &validators.etag { + request = request.header(IF_NONE_MATCH, etag); + } + if let Some(last_modified) = &validators.last_modified { + request = request.header(IF_MODIFIED_SINCE, last_modified); + } + + let response = request.send()?; + if response.status() == StatusCode::NOT_MODIFIED { + return Ok(FetchResult::NotModified); + } + + let response = response.error_for_status()?; + let header_str = |name| { + response + .headers() + .get(name) + .and_then(|v| v.to_str().ok()) + .map(str::to_string) + }; + let validators = Validators { + etag: header_str(ETAG), + last_modified: header_str(LAST_MODIFIED), + }; + let body = response.text()?; + Ok(FetchResult::Modified { body, validators }) } /// Fetches without treating any outcome as fatal: a non-2xx status, or a @@ -59,6 +105,78 @@ mod tests { format!("http://127.0.0.1:{port}/") } + /// Like `serve_once`, but also hands back everything the client sent, + /// lowercased, so a test can assert on which headers were present. + fn serve_once_capturing(response: &'static str) -> (String, std::sync::mpsc::Receiver) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).unwrap_or(0); + let _ = tx.send(String::from_utf8_lossy(&buf[..n]).to_lowercase()); + stream.write_all(response.as_bytes()).unwrap(); + }); + (format!("http://127.0.0.1:{port}/"), rx) + } + + #[test] + fn fetch_sends_validators_as_conditional_headers() { + let (url, requests) = + serve_once_capturing("HTTP/1.1 304 Not Modified\r\nConnection: close\r\n\r\n"); + let validators = Validators { + etag: Some("\"abc\"".to_string()), + last_modified: Some("Sun, 06 Sep 2026 00:00:00 GMT".to_string()), + }; + + let result = fetch(&url, &validators).unwrap(); + + assert!(matches!(result, FetchResult::NotModified)); + let request = requests.recv().unwrap(); + assert!(request.contains("if-none-match: \"abc\"")); + assert!(request.contains("if-modified-since: sun, 06 sep 2026 00:00:00 gmt")); + } + + #[test] + fn fetch_omits_conditional_headers_when_no_validators_are_cached() { + let (url, requests) = serve_once_capturing( + "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", + ); + + fetch(&url, &Validators::default()).unwrap(); + + let request = requests.recv().unwrap(); + assert!(!request.contains("if-none-match")); + assert!(!request.contains("if-modified-since")); + } + + #[test] + fn fetch_returns_modified_body_and_new_validators_on_200() { + let url = serve_once(concat!( + "HTTP/1.1 200 OK\r\n", + "ETag: \"xyz\"\r\n", + "Last-Modified: Mon, 07 Sep 2026 00:00:00 GMT\r\n", + "Content-Length: 5\r\n", + "Connection: close\r\n\r\n", + "hello" + )); + + let result = fetch(&url, &Validators::default()).unwrap(); + + match result { + FetchResult::Modified { body, validators } => { + assert_eq!(body, "hello"); + assert_eq!(validators.etag.as_deref(), Some("\"xyz\"")); + assert_eq!( + validators.last_modified.as_deref(), + Some("Mon, 07 Sep 2026 00:00:00 GMT") + ); + } + FetchResult::NotModified => panic!("expected a Modified result"), + } + } + #[test] fn fetch_status_returns_status_and_body() { let url = diff --git a/src/opml.rs b/src/opml.rs index 72062e1..13beb6e 100644 --- a/src/opml.rs +++ b/src/opml.rs @@ -38,6 +38,7 @@ mod tests { description: "".into(), method_source: "".into(), method_label: "example".into(), + interval: std::time::Duration::from_secs(300), } } diff --git a/src/rss.rs b/src/rss.rs index fa694e7..e5b9924 100644 --- a/src/rss.rs +++ b/src/rss.rs @@ -81,6 +81,7 @@ mod tests { description: "An example feed".into(), method_source: "".into(), method_label: "example".into(), + interval: std::time::Duration::from_secs(300), } } diff --git a/src/scheduler.rs b/src/scheduler.rs index 8538550..3d1fad4 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -3,28 +3,87 @@ use crate::config::{Config, SiteConfig}; use crate::http_fetch; use crate::rss::render_channel; use crate::run_method::run_method; +use std::collections::HashSet; use std::path::Path; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; -/// Cycles through configured sites forever, dispatching one fetch per site -/// on its own worker thread and sleeping `config.delay` between dispatches. -/// There is no per-site interval: `delay` is the only timing knob, so a -/// site's effective refetch period is `delay * config.sites.len()`. The -/// coordinator never waits on a dispatched fetch, so a slow site can't stall -/// the schedule for the rest. +/// How often the coordinator wakes to check which sites are due. +const TICK: Duration = Duration::from_secs(1); + +/// Minimum spacing between dispatches issued within the same tick, so sites +/// that become due at the same moment don't all get hit at once. +const DISPATCH_STAGGER: Duration = Duration::from_secs(2); + +/// Tracks, per site (by index into `config.sites`), whether a fetch is +/// currently in flight and when the last attempt was made -- the basis for +/// deciding whether a site is due. Kept in-process rather than read back +/// from the feed cache so a failing site backs off by its own `interval` +/// instead of being retried every tick. +struct Schedule { + last_attempt: Vec>>, + in_flight: Mutex>, +} + +impl Schedule { + fn new(site_count: usize) -> Self { + Self { + last_attempt: (0..site_count).map(|_| Mutex::new(None)).collect(), + in_flight: Mutex::new(HashSet::new()), + } + } + + fn is_due(&self, index: usize, interval: Duration) -> bool { + match *self.last_attempt[index].lock().unwrap() { + None => true, + Some(last) => last.elapsed() >= interval, + } + } + + /// Claims `index` for dispatch, unless a fetch for it is already in + /// flight -- which keeps a slow site from being dispatched again before + /// its previous fetch finishes. Records the attempt time on success so + /// `is_due` won't fire again until `interval` elapses. + fn try_claim(&self, index: usize) -> bool { + let mut in_flight = self.in_flight.lock().unwrap(); + if !in_flight.insert(index) { + return false; + } + *self.last_attempt[index].lock().unwrap() = Some(Instant::now()); + true + } + + fn release(&self, index: usize) { + self.in_flight.lock().unwrap().remove(&index); + } +} + +/// Wakes every `TICK` and dispatches any site whose own `interval` (see +/// `SiteConfig::interval`) has elapsed since its last attempt, each on its +/// own worker thread. Sites due in the same tick are staggered by +/// `DISPATCH_STAGGER` rather than fired all at once. The coordinator never +/// waits on a dispatched fetch, so a slow site can't stall the schedule for +/// the rest -- it's simply skipped on later ticks until it finishes. pub fn run(config: Arc) -> ! { + let schedule = Arc::new(Schedule::new(config.sites.len())); loop { - if config.sites.is_empty() { - thread::sleep(Duration::from_secs(1)); - continue; - } for index in 0..config.sites.len() { + if !schedule.is_due(index, config.sites[index].interval) { + continue; + } + if !schedule.try_claim(index) { + continue; + } let worker_config = Arc::clone(&config); - thread::spawn(move || fetch_one(&worker_config.sites[index], &worker_config.state_dir)); - thread::sleep(config.delay); + let worker_schedule = Arc::clone(&schedule); + thread::spawn(move || { + fetch_one(&worker_config.sites[index], &worker_config.state_dir); + worker_schedule.release(index); + }); + thread::sleep(DISPATCH_STAGGER); } + thread::sleep(TICK); } } @@ -42,8 +101,13 @@ pub fn poll_all(config: &Config) { } fn fetch_one(site: &SiteConfig, state_dir: &Path) { - let html = match http_fetch::fetch(&site.url) { - Ok(html) => html, + let validators = cache::read_validators(state_dir, &site.name); + let (html, new_validators) = match http_fetch::fetch(&site.url, &validators) { + Ok(http_fetch::FetchResult::NotModified) => { + tracing::debug!(site = %site.name, "not modified; keeping cached feed"); + return; + } + Ok(http_fetch::FetchResult::Modified { body, validators }) => (body, validators), Err(e) => { tracing::error!(site = %site.name, error = %e, "fetch failed; keeping cached feed"); return; @@ -62,4 +126,10 @@ fn fetch_one(site: &SiteConfig, state_dir: &Path) { if let Err(e) = cache::write_feed(state_dir, &site.name, &xml) { tracing::error!(site = %site.name, error = %e, "failed to write feed cache"); } + // Only persisted once the fetch has actually been turned into a feed -- + // a failed method script leaves the old validators in place, so a fixed + // script gets a real re-fetch to work with rather than a cached 304. + if let Err(e) = cache::write_validators(state_dir, &site.name, &new_validators) { + tracing::error!(site = %site.name, error = %e, "failed to write fetch validators"); + } } -- knotmirror 0.0.1