From d96b4b7b407c102313247bd00fb3b0cd8f5c5f02 Mon Sep 17 00:00:00 2001 From: forkwright Date: Thu, 2 Jul 2026 09:53:44 -0500 Subject: [PATCH] fix(komide): bound feed/episode fetch, fix backoff overflow, add coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - #368: fetch_feed/download_episode enforce configurable caps (max_feed_bytes default 20 MiB, max_episode_bytes default 1 GiB) — Content-Length precheck plus per-chunk streaming total, aborting with ResponseTooLarge; download_episode streams to disk and cleans up a partial file on failure. - #369: exponential backoff clamps the exponent (min 63) and uses saturating arithmetic, so it no longer overflows after 64 failures. - #455: fetch_feed/download_episode call error_for_status after the 304 early-return, so 4xx/5xx bodies are no longer treated as content. - #370/#371/#456: coverage for the streaming episode download, the refresh_feed poll path, and the NotModified (304)/store_validators path, using an in-crate one-shot HTTP test server (no new workspace dep). Closes #368 Closes #369 Closes #370 Closes #371 Closes #455 Closes #456 Gate-Passed: kanon 0.1.5 +stages:fmt,check,clippy,nextest,lint sha:1dad62d4d5fc715b1481b59b3aaa6e9cd966824b --- Cargo.lock | 1 + crates/horismos/src/lib.rs | 9 + crates/horismos/src/subsystems.rs | 6 + crates/komide/Cargo.toml | 1 + crates/komide/src/error.rs | 8 + crates/komide/src/fetch.rs | 338 +++++++++++++++++++++++++++-- crates/komide/src/lib.rs | 1 + crates/komide/src/scheduler.rs | 50 ++++- crates/komide/src/service/mod.rs | 39 +++- crates/komide/src/service/tests.rs | 270 ++++++++++++++++++++++- crates/komide/src/test_support.rs | 80 +++++++ 11 files changed, 762 insertions(+), 41 deletions(-) create mode 100644 crates/komide/src/test_support.rs diff --git a/Cargo.lock b/Cargo.lock index 467f7272..1aa891f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2639,6 +2639,7 @@ dependencies = [ "serde", "snafu", "sqlx", + "tempfile", "themelion", "tokio", "tracing", diff --git a/crates/horismos/src/lib.rs b/crates/horismos/src/lib.rs index 96fd3ff8..cd16fa76 100644 --- a/crates/horismos/src/lib.rs +++ b/crates/horismos/src/lib.rs @@ -89,6 +89,15 @@ mod tests { assert_eq!(config.kritike.scan_interval_hours, 24); } + #[test] + fn default_komide_config_has_correct_values() { + let config = Config::default(); + assert_eq!(config.komide.max_feed_bytes, 20 * 1024 * 1024); + assert_eq!(config.komide.max_episode_bytes, 1024 * 1024 * 1024); + assert_eq!(config.komide.max_backoff_minutes, 240); + assert_eq!(config.komide.fetch_timeout_secs, 30); + } + #[test] fn default_syndesmos_config_has_correct_values() { let config = Config::default(); diff --git a/crates/horismos/src/subsystems.rs b/crates/horismos/src/subsystems.rs index 13dbf52d..febe3986 100644 --- a/crates/horismos/src/subsystems.rs +++ b/crates/horismos/src/subsystems.rs @@ -403,6 +403,10 @@ pub struct KomideConfig { pub auto_download_latest_n: u64, /// Request timeout for feed fetches in seconds. pub fetch_timeout_secs: u64, + /// Maximum feed response body size in bytes; larger responses are rejected. + pub max_feed_bytes: u64, + /// Maximum episode download size in bytes; larger downloads are rejected. + pub max_episode_bytes: u64, /// Maximum exponential-backoff window (minutes) between feed polls after /// consecutive failures. pub max_backoff_minutes: u64, @@ -421,6 +425,8 @@ impl Default for KomideConfig { news_retention_articles: 500, auto_download_latest_n: 3, fetch_timeout_secs: 30, + max_feed_bytes: 20 * 1024 * 1024, + max_episode_bytes: 1024 * 1024 * 1024, max_backoff_minutes: 240, jitter_percent: 10.0, } diff --git a/crates/komide/Cargo.toml b/crates/komide/Cargo.toml index 50283bef..0dc6a694 100644 --- a/crates/komide/Cargo.toml +++ b/crates/komide/Cargo.toml @@ -22,6 +22,7 @@ jiff = { workspace = true } rstest = { workspace = true } tokio = { workspace = true, features = ["full"] } sqlx = { workspace = true } +tempfile = "3" [package.metadata.kanon] maturity = "alpha" diff --git a/crates/komide/src/error.rs b/crates/komide/src/error.rs index fbb06dba..4372098f 100644 --- a/crates/komide/src/error.rs +++ b/crates/komide/src/error.rs @@ -36,6 +36,14 @@ pub enum KomideError { location: snafu::Location, }, + #[snafu(display("response body for {url} exceeds the {limit}-byte cap"))] + ResponseTooLarge { + url: String, + limit: u64, + #[snafu(implicit)] + location: snafu::Location, + }, + #[snafu(display("invalid feed URL: {url}"))] InvalidUrl { url: String, diff --git a/crates/komide/src/fetch.rs b/crates/komide/src/fetch.rs index d70511ec..22d76dbb 100644 --- a/crates/komide/src/fetch.rs +++ b/crates/komide/src/fetch.rs @@ -1,7 +1,9 @@ -use reqwest::{Client, StatusCode, header}; -use snafu::ResultExt; +use reqwest::{Client, Response, StatusCode, header}; +use snafu::{ResultExt, ensure}; -use crate::error::{EpisodeDownloadSnafu, EpisodeIoSnafu, FeedFetchSnafu, KomideError}; +use crate::error::{ + EpisodeDownloadSnafu, EpisodeIoSnafu, FeedFetchSnafu, KomideError, ResponseTooLargeSnafu, +}; #[non_exhaustive] pub enum FetchResult { @@ -16,12 +18,15 @@ pub enum FetchResult { /// Fetch a feed URL using conditional GET if ETag or Last-Modified is provided. /// /// Returns `FetchResult::NotModified` on HTTP 304, or `FetchResult::Content` -/// with the response body and freshly received cache validators. +/// with the response body and freshly received cache validators. Non-success +/// statuses are errors; the body is streamed and rejected once it exceeds +/// `max_bytes`. pub async fn fetch_feed( client: &Client, url: &str, etag: Option<&str>, last_modified: Option<&str>, + max_bytes: u64, ) -> Result { let mut req = client.get(url); @@ -40,6 +45,11 @@ pub async fn fetch_feed( return Ok(FetchResult::NotModified); } + // WHY: a 4xx/5xx body is an error page, not feed content; reject before reading. + let response = response.error_for_status().context(FeedFetchSnafu { + url: url.to_string(), + })?; + let new_etag = response .headers() .get(header::ETAG) @@ -52,13 +62,7 @@ pub async fn fetch_feed( .and_then(|v| v.to_str().ok()) .map(str::to_owned); - let bytes = response - .bytes() - .await - .context(FeedFetchSnafu { - url: url.to_string(), - })? - .to_vec(); + let bytes = read_body_capped(response, url, max_bytes).await?; Ok(FetchResult::Content { bytes, @@ -67,40 +71,127 @@ pub async fn fetch_feed( }) } -/// Download episode audio to the given path. Returns file size in bytes. +/// Stream a feed response body, failing with `ResponseTooLarge` once the +/// declared or accumulated size exceeds `max_bytes`. +pub(crate) async fn read_body_capped( + mut response: Response, + url: &str, + max_bytes: u64, +) -> Result, KomideError> { + if let Some(declared) = response.content_length() { + ensure!( + declared <= max_bytes, + ResponseTooLargeSnafu { + url: url.to_string(), + limit: max_bytes, + } + ); + } + + let mut bytes: Vec = Vec::new(); + while let Some(chunk) = response.chunk().await.context(FeedFetchSnafu { + url: url.to_string(), + })? { + let total = (bytes.len() as u64).saturating_add(chunk.len() as u64); + ensure!( + total <= max_bytes, + ResponseTooLargeSnafu { + url: url.to_string(), + limit: max_bytes, + } + ); + bytes.extend_from_slice(&chunk); + } + + Ok(bytes) +} + +/// Download episode audio to the given path, streaming chunks straight to +/// disk. Returns the file size in bytes. Downloads whose declared or streamed +/// size exceeds `max_bytes` abort with `ResponseTooLarge`, and any partially +/// written file is removed. pub async fn download_episode( client: &Client, url: &str, dest: &std::path::Path, + max_bytes: u64, ) -> Result { - use tokio::io::AsyncWriteExt; - let response = client.get(url).send().await.context(EpisodeDownloadSnafu { url: url.to_string(), })?; - let bytes = response.bytes().await.context(EpisodeDownloadSnafu { + // WHY: a 4xx/5xx body is an error page, not audio; reject before touching disk. + let response = response.error_for_status().context(EpisodeDownloadSnafu { url: url.to_string(), })?; + if let Some(declared) = response.content_length() { + ensure!( + declared <= max_bytes, + ResponseTooLargeSnafu { + url: url.to_string(), + limit: max_bytes, + } + ); + } + let path_str = dest.display().to_string(); - let mut file = tokio::fs::File::create(dest) + let file = tokio::fs::File::create(dest) .await .context(EpisodeIoSnafu { path: path_str.clone(), })?; - file.write_all(&bytes) - .await - .context(EpisodeIoSnafu { path: path_str })?; + match stream_body_to_file(response, file, url, &path_str, max_bytes).await { + Ok(written) => Ok(written), + Err(err) => { + // WHY: never leave a truncated or over-cap partial file on disk. + tokio::fs::remove_file(dest).await.ok(); + Err(err) + } + } +} - Ok(bytes.len() as u64) +async fn stream_body_to_file( + mut response: Response, + mut file: tokio::fs::File, + url: &str, + path: &str, + max_bytes: u64, +) -> Result { + use tokio::io::AsyncWriteExt; + + let mut written: u64 = 0; + while let Some(chunk) = response.chunk().await.context(EpisodeDownloadSnafu { + url: url.to_string(), + })? { + written = written.saturating_add(chunk.len() as u64); + ensure!( + written <= max_bytes, + ResponseTooLargeSnafu { + url: url.to_string(), + limit: max_bytes, + } + ); + file.write_all(&chunk).await.context(EpisodeIoSnafu { + path: path.to_string(), + })?; + } + + file.flush().await.context(EpisodeIoSnafu { + path: path.to_string(), + })?; + + Ok(written) } #[cfg(test)] mod tests { use super::*; + use crate::test_support::{http_response, http_response_close_delimited, spawn_scripted_http}; + + const CAP: u64 = 1024; #[test] fn fetch_result_not_modified_variant() { @@ -143,4 +234,211 @@ mod tests { FetchResult::NotModified => panic!("expected Content"), } } + + #[tokio::test] + async fn fetch_feed_returns_content_and_captures_validators() { + let (url, handle) = spawn_scripted_http(vec![http_response( + 200, + "OK", + &[ + ("etag", "\"v1\""), + ("last-modified", "Wed, 01 Jan 2026 00:00:00 GMT"), + ], + b"", + )]) + .await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await.unwrap(); + match result { + FetchResult::Content { + bytes, + etag, + last_modified, + } => { + assert_eq!(bytes, b""); + assert_eq!(etag.as_deref(), Some("\"v1\"")); + assert_eq!( + last_modified.as_deref(), + Some("Wed, 01 Jan 2026 00:00:00 GMT") + ); + } + FetchResult::NotModified => panic!("expected Content"), + } + + let requests = handle.await.unwrap(); + let head = requests[0].to_lowercase(); + assert!( + !head.contains("if-none-match"), + "unconditional fetch must not send validators" + ); + } + + #[tokio::test] + async fn fetch_feed_304_returns_not_modified_and_sends_validators() { + // Regression guard for the status-check ORDER: 304 must short-circuit + // to NotModified and never be treated as content or as an error. + let (url, handle) = + spawn_scripted_http(vec![http_response(304, "Not Modified", &[], b"")]).await; + + let client = Client::new(); + let result = fetch_feed( + &client, + &url, + Some("\"v1\""), + Some("Wed, 01 Jan 2026 00:00:00 GMT"), + CAP, + ) + .await + .unwrap(); + assert!(matches!(result, FetchResult::NotModified)); + + let requests = handle.await.unwrap(); + let head = requests[0].to_lowercase(); + assert!(head.contains("if-none-match: \"v1\"")); + assert!(head.contains("if-modified-since: wed, 01 jan 2026 00:00:00 gmt")); + } + + #[tokio::test] + async fn fetch_feed_500_returns_error_not_content() { + // The 500 body parses as a feed; it must still be rejected on status. + let (url, _handle) = spawn_scripted_http(vec![http_response( + 500, + "Internal Server Error", + &[], + b"t", + )]) + .await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await; + assert!(matches!(result, Err(KomideError::FeedFetch { .. }))); + } + + #[tokio::test] + async fn fetch_feed_404_returns_error() { + let (url, _handle) = + spawn_scripted_http(vec![http_response(404, "Not Found", &[], b"missing")]).await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await; + assert!(matches!(result, Err(KomideError::FeedFetch { .. }))); + } + + #[tokio::test] + async fn fetch_feed_declared_over_cap_rejected_up_front() { + let body = vec![b'x'; (CAP as usize) * 2]; + let (url, _handle) = spawn_scripted_http(vec![http_response(200, "OK", &[], &body)]).await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await; + assert!(matches!( + result, + Err(KomideError::ResponseTooLarge { limit, .. }) if limit == CAP + )); + } + + #[tokio::test] + async fn fetch_feed_streamed_over_cap_rejected_without_content_length() { + // No Content-Length header: the cap must trip in the streaming loop. + let body = vec![b'x'; (CAP as usize) * 4]; + let (url, _handle) = + spawn_scripted_http(vec![http_response_close_delimited(200, "OK", &body)]).await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await; + assert!(matches!(result, Err(KomideError::ResponseTooLarge { .. }))); + } + + #[tokio::test] + async fn fetch_feed_body_exactly_at_cap_succeeds() { + let body = vec![b'x'; CAP as usize]; + let (url, _handle) = + spawn_scripted_http(vec![http_response_close_delimited(200, "OK", &body)]).await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await.unwrap(); + match result { + FetchResult::Content { bytes, .. } => assert_eq!(bytes.len(), CAP as usize), + FetchResult::NotModified => panic!("expected Content"), + } + } + + #[tokio::test] + async fn fetch_feed_body_one_byte_over_cap_fails() { + let body = vec![b'x'; (CAP as usize) + 1]; + let (url, _handle) = + spawn_scripted_http(vec![http_response_close_delimited(200, "OK", &body)]).await; + + let client = Client::new(); + let result = fetch_feed(&client, &url, None, None, CAP).await; + assert!(matches!(result, Err(KomideError::ResponseTooLarge { .. }))); + } + + #[tokio::test] + async fn download_episode_writes_expected_content_and_size() { + let body = b"pretend-this-is-audio-bytes"; + let (url, _handle) = spawn_scripted_http(vec![http_response(200, "OK", &[], body)]).await; + + let dir = tempfile::tempdir().unwrap(); + let dest = dir.path().join("episode.mp3"); + let client = Client::new(); + + let written = download_episode(&client, &url, &dest, CAP).await.unwrap(); + assert_eq!(written, body.len() as u64); + + let on_disk = std::fs::read(&dest).unwrap(); + assert_eq!(on_disk, body); + assert_eq!(std::fs::metadata(&dest).unwrap().len(), written); + } + + #[tokio::test] + async fn download_episode_500_returns_error_and_writes_no_file() { + let (url, _handle) = spawn_scripted_http(vec![http_response( + 500, + "Internal Server Error", + &[], + b"oops", + )]) + .await; + + let dir = tempfile::tempdir().unwrap(); + let dest = dir.path().join("episode.mp3"); + let client = Client::new(); + + let result = download_episode(&client, &url, &dest, CAP).await; + assert!(matches!(result, Err(KomideError::EpisodeDownload { .. }))); + assert!(!dest.exists(), "no file may be created on HTTP error"); + } + + #[tokio::test] + async fn download_episode_declared_over_cap_fails_before_creating_file() { + let body = vec![b'x'; (CAP as usize) * 2]; + let (url, _handle) = spawn_scripted_http(vec![http_response(200, "OK", &[], &body)]).await; + + let dir = tempfile::tempdir().unwrap(); + let dest = dir.path().join("episode.mp3"); + let client = Client::new(); + + let result = download_episode(&client, &url, &dest, CAP).await; + assert!(matches!(result, Err(KomideError::ResponseTooLarge { .. }))); + assert!(!dest.exists(), "over-cap download must not create the file"); + } + + #[tokio::test] + async fn download_episode_streamed_over_cap_removes_partial_file() { + // No Content-Length header: the cap trips mid-stream, after bytes may + // already be on disk; the partial file must be removed. + let body = vec![b'x'; (CAP as usize) * 4]; + let (url, _handle) = + spawn_scripted_http(vec![http_response_close_delimited(200, "OK", &body)]).await; + + let dir = tempfile::tempdir().unwrap(); + let dest = dir.path().join("episode.mp3"); + let client = Client::new(); + + let result = download_episode(&client, &url, &dest, CAP).await; + assert!(matches!(result, Err(KomideError::ResponseTooLarge { .. }))); + assert!(!dest.exists(), "partial over-cap file must be removed"); + } } diff --git a/crates/komide/src/lib.rs b/crates/komide/src/lib.rs index 7879ed8b..6b79c772 100644 --- a/crates/komide/src/lib.rs +++ b/crates/komide/src/lib.rs @@ -5,6 +5,7 @@ pub mod parser; pub mod podcast; pub mod scheduler; pub mod service; +pub(crate) mod test_support; pub use error::KomideError; pub use service::{FeedRefreshResult, FeedSchedulerService, FeedSummary}; diff --git a/crates/komide/src/scheduler.rs b/crates/komide/src/scheduler.rs index 4ac47514..9bcb0260 100644 --- a/crates/komide/src/scheduler.rs +++ b/crates/komide/src/scheduler.rs @@ -36,7 +36,11 @@ impl FeedState { if self.failure_count == 0 { return self.base_interval_minutes; } - let backed_off = self.base_interval_minutes * 2u64.pow(self.failure_count); + // WHY: clamp the shift below u64::BITS and saturate the multiply so a + // long outage (64+ consecutive failures) cannot overflow and collapse + // the interval back to a tiny value. + let doubling = 1u64 << self.failure_count.min(63); + let backed_off = self.base_interval_minutes.saturating_mul(doubling); backed_off.min(self.max_backoff_minutes) } @@ -218,6 +222,50 @@ mod tests { ); } + #[test] + fn backoff_survives_failure_count_100() { + // Regression guard: 2^100 used to overflow (panic in debug, wrap to a + // short interval in release); the clamp must hold the configured max. + let mut state = FeedState::new(30, &test_config()); + state.failure_count = 100; + assert_eq!( + state.current_interval_minutes(), + KomideConfig::default().max_backoff_minutes + ); + } + + #[test] + fn backoff_at_shift_clamp_boundary_stays_capped() { + let mut state = FeedState::new(30, &test_config()); + for failures in [62, 63, 64, 65, u32::MAX] { + state.failure_count = failures; + assert_eq!( + state.current_interval_minutes(), + KomideConfig::default().max_backoff_minutes, + "interval must stay at the cap for {failures} failures" + ); + } + } + + #[test] + fn backoff_never_zero_and_monotonic_up_to_cap() { + let mut state = FeedState::new(30, &test_config()); + let mut previous = 0u64; + for failures in 0..=200u32 { + state.failure_count = failures; + let interval = state.current_interval_minutes(); + assert!( + interval >= 1, + "interval collapsed to zero at {failures} failures" + ); + assert!( + interval >= previous, + "interval regressed at {failures} failures: {interval} < {previous}" + ); + previous = interval; + } + } + #[test] fn custom_max_backoff_observed() { // WHY: non-default config must observably cap backoff sooner. diff --git a/crates/komide/src/service/mod.rs b/crates/komide/src/service/mod.rs index 08be9912..dfd243d3 100644 --- a/crates/komide/src/service/mod.rs +++ b/crates/komide/src/service/mod.rs @@ -77,7 +77,7 @@ impl FeedSchedulerService { } // Fetch and parse to populate metadata - let feed_bytes = fetch_bytes(&self.client, url).await?; + let feed_bytes = fetch_bytes(&self.client, url, self.config.max_feed_bytes).await?; let parsed = parse_feed(&feed_bytes)?; let feed_id = FeedId::new(); @@ -130,7 +130,7 @@ impl FeedSchedulerService { return Ok(bytes_to_feed_id(&existing.id)); } - let feed_bytes = fetch_bytes(&self.client, url).await?; + let feed_bytes = fetch_bytes(&self.client, url, self.config.max_feed_bytes).await?; let parsed = parse_feed(&feed_bytes)?; let feed_id = FeedId::new(); @@ -314,8 +314,14 @@ impl FeedSchedulerService { let url = &sub.feed_url; let (etag, last_modified) = self.cached_validators(url).await; - let fetch_result = - fetch_feed(&self.client, url, etag.as_deref(), last_modified.as_deref()).await?; + let fetch_result = fetch_feed( + &self.client, + url, + etag.as_deref(), + last_modified.as_deref(), + self.config.max_feed_bytes, + ) + .await?; match fetch_result { FetchResult::NotModified => { @@ -375,8 +381,14 @@ impl FeedSchedulerService { let url = &feed.url; let (etag, last_modified) = self.cached_validators(url).await; - let fetch_result = - fetch_feed(&self.client, url, etag.as_deref(), last_modified.as_deref()).await?; + let fetch_result = fetch_feed( + &self.client, + url, + etag.as_deref(), + last_modified.as_deref(), + self.config.max_feed_bytes, + ) + .await?; match fetch_result { FetchResult::NotModified => { @@ -586,21 +598,24 @@ fn validate_url(input: &str) -> Result<(), KomideError> { } } -async fn fetch_bytes(client: &reqwest::Client, url: &str) -> Result, KomideError> { +async fn fetch_bytes( + client: &reqwest::Client, + url: &str, + max_bytes: u64, +) -> Result, KomideError> { use crate::error::FeedFetchSnafu; - client + let response = client .get(url) .send() .await .context(FeedFetchSnafu { url: url.to_string(), })? - .bytes() - .await + .error_for_status() .context(FeedFetchSnafu { url: url.to_string(), - }) - .map(|b| b.to_vec()) + })?; + crate::fetch::read_body_capped(response, url, max_bytes).await } pub(crate) fn bytes_to_feed_id(bytes: &[u8]) -> FeedId { diff --git a/crates/komide/src/service/tests.rs b/crates/komide/src/service/tests.rs index 101cee10..9981c99f 100644 --- a/crates/komide/src/service/tests.rs +++ b/crates/komide/src/service/tests.rs @@ -1,9 +1,57 @@ use apotheke::DbPools; use apotheke::migrate::MIGRATOR; use sqlx::SqlitePool; -use themelion::aggelia::create_event_bus; +use themelion::aggelia::{HarmoniaEvent, create_event_bus}; use super::*; +use crate::test_support::{http_response, spawn_scripted_http}; + +const RSS_TWO_EPISODES: &[u8] = br#" + + + Test Podcast + A test podcast feed + + Episode 1 + ep-001 + Mon, 01 Jan 2024 00:00:00 +0000 + + + + Episode 2 + ep-002 + Tue, 02 Jan 2024 00:00:00 +0000 + + + +"#; + +/// Atom fixture with two fresh articles. Published timestamps are "now" so +/// the default 30-day retention pass keeps them. +fn atom_two_articles() -> Vec { + let now = now_iso8601(); + format!( + r#" + + Test News + + article-001 + Breaking News + {now} + Something happened + + + + article-002 + Follow Up + {now} + More details + + +"# + ) + .into_bytes() +} async fn setup() -> (FeedSchedulerService, themelion::aggelia::EventReceiver) { let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); @@ -104,7 +152,7 @@ async fn mark_consumed_nonexistent_is_ok() { #[tokio::test] async fn insert_episodes_deduplicates_by_guid() { let (svc, _rx) = setup().await; - let sub_id = make_subscription(&svc).await; + let sub_id = make_subscription(&svc, "https://example.com/podcast.xml").await; let entries = vec![ make_podcast_entry("ep-001", "Episode 1"), @@ -127,7 +175,7 @@ async fn insert_episodes_deduplicates_by_guid() { #[tokio::test] async fn insert_articles_deduplicates_by_guid() { let (svc, _rx) = setup().await; - let feed_id = make_news_feed(&svc).await; + let feed_id = make_news_feed(&svc, "https://example.com/news.xml").await; let entries = vec![ make_news_entry("art-001", "Article 1"), @@ -153,7 +201,7 @@ async fn episode_available_event_emitted_on_new_episode() { let (tx, mut rx) = create_event_bus(64); let svc = FeedSchedulerService::new(db, tx, reqwest::Client::new(), KomideConfig::default()); - let sub_id = make_subscription(&svc).await; + let sub_id = make_subscription(&svc, "https://example.com/podcast.xml").await; let entries = vec![make_podcast_entry("ep-new", "New Episode")]; let now = now_iso8601(); svc.insert_new_podcast_episodes(&sub_id, &entries, &now) @@ -167,14 +215,220 @@ async fn episode_available_event_emitted_on_new_episode() { )); } +// ── refresh_feed integration (scripted in-process HTTP server) ─────────── + +#[tokio::test] +async fn refresh_podcast_feed_inserts_items_and_emits_event() { + let (svc, mut rx) = setup().await; + let (url, _handle) = spawn_scripted_http(vec![http_response( + 200, + "OK", + &[("etag", "\"v1\"")], + RSS_TWO_EPISODES, + )]) + .await; + let sub_id = make_subscription(&svc, &url).await; + let feed_id = bytes_to_feed_id(&sub_id); + + let result = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(result.new_items, 2); + assert_eq!(result.total_items, 2); + assert_eq!(result.feed_id, feed_id); + + let sub = podcast::get_subscription(&svc.db.read, &sub_id) + .await + .unwrap() + .unwrap(); + assert!( + sub.last_checked_at.is_some(), + "refresh must update last_checked_at" + ); + + let mut saw_refreshed = false; + while let Ok(event) = rx.try_recv() { + if matches!( + event, + HarmoniaEvent::FeedRefreshed { + new_items: 2, + media_type: MediaType::Podcast, + .. + } + ) { + saw_refreshed = true; + } + } + assert!(saw_refreshed, "FeedRefreshed event must be emitted"); +} + +#[tokio::test] +async fn refresh_news_feed_inserts_articles_and_emits_event() { + let (svc, mut rx) = setup().await; + let (url, _handle) = + spawn_scripted_http(vec![http_response(200, "OK", &[], &atom_two_articles())]).await; + let feed_bytes = make_news_feed(&svc, &url).await; + let feed_id = bytes_to_feed_id(&feed_bytes); + + let result = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(result.new_items, 2); + assert_eq!(result.total_items, 2); + + let feed = news::get_feed(&svc.db.read, &feed_bytes) + .await + .unwrap() + .unwrap(); + assert!( + feed.last_fetched_at.is_some(), + "refresh must update last_fetched_at" + ); + + let mut saw_refreshed = false; + while let Ok(event) = rx.try_recv() { + if matches!( + event, + HarmoniaEvent::FeedRefreshed { + new_items: 2, + media_type: MediaType::News, + .. + } + ) { + saw_refreshed = true; + } + } + assert!(saw_refreshed, "FeedRefreshed event must be emitted"); +} + +#[tokio::test] +async fn refresh_feed_unknown_id_returns_feed_not_found() { + let (svc, _rx) = setup().await; + let result = svc.refresh_feed(FeedId::new()).await; + assert!(matches!(result, Err(KomideError::FeedNotFound { .. }))); +} + +#[tokio::test] +async fn refresh_feed_http_500_is_error_and_leaves_db_untouched() { + let (svc, _rx) = setup().await; + let (url, _handle) = spawn_scripted_http(vec![http_response( + 500, + "Internal Server Error", + &[], + b"oops", + )]) + .await; + let sub_id = make_subscription(&svc, &url).await; + + let result = svc.refresh_feed(bytes_to_feed_id(&sub_id)).await; + assert!(matches!(result, Err(KomideError::FeedFetch { .. }))); + + let episodes = podcast::list_episodes(&svc.db.read, &sub_id, 10, 0) + .await + .unwrap(); + assert!(episodes.is_empty(), "a 500 must not insert episodes"); + + let sub = podcast::get_subscription(&svc.db.read, &sub_id) + .await + .unwrap() + .unwrap(); + assert!( + sub.last_checked_at.is_none(), + "a 500 must not update last_checked_at" + ); +} + +#[tokio::test] +async fn refresh_podcast_feed_second_call_with_304_returns_zero_new_items() { + let (svc, _rx) = setup().await; + let (url, handle) = spawn_scripted_http(vec![ + http_response(200, "OK", &[("etag", "\"v1\"")], RSS_TWO_EPISODES), + http_response(304, "Not Modified", &[], b""), + ]) + .await; + let sub_id = make_subscription(&svc, &url).await; + let feed_id = bytes_to_feed_id(&sub_id); + + let first = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(first.new_items, 2); + + let second = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(second.new_items, 0); + assert_eq!( + second.total_items, 2, + "total_items on 304 is the stored episode count" + ); + + let requests = handle.await.unwrap(); + assert!( + !requests[0].to_lowercase().contains("if-none-match"), + "first request must be unconditional" + ); + assert!( + requests[1].to_lowercase().contains("if-none-match: \"v1\""), + "second request must forward the stored ETag" + ); +} + +#[tokio::test] +async fn refresh_news_feed_second_call_with_304_returns_zero_new_items() { + let (svc, _rx) = setup().await; + let (url, handle) = spawn_scripted_http(vec![ + http_response(200, "OK", &[("etag", "\"n1\"")], &atom_two_articles()), + http_response(304, "Not Modified", &[], b""), + ]) + .await; + let feed_bytes = make_news_feed(&svc, &url).await; + let feed_id = bytes_to_feed_id(&feed_bytes); + + let first = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(first.new_items, 2); + + let second = svc.refresh_feed(feed_id).await.unwrap(); + assert_eq!(second.new_items, 0); + assert_eq!( + second.total_items, 2, + "total_items on 304 is the stored article count" + ); + + let requests = handle.await.unwrap(); + assert!( + requests[1].to_lowercase().contains("if-none-match: \"n1\""), + "second request must forward the stored ETag" + ); +} + +#[tokio::test] +async fn store_validators_ignores_empty_pair_and_keeps_nonempty() { + let (svc, _rx) = setup().await; + let url = "https://example.com/feed.xml"; + + svc.store_validators(url, None, None).await; + assert_eq!( + svc.cached_validators(url).await, + (None, None), + "an empty validator pair must not be cached" + ); + + svc.store_validators(url, Some("\"e1\"".to_string()), None) + .await; + assert_eq!( + svc.cached_validators(url).await, + (Some("\"e1\"".to_string()), None) + ); + + // A later empty pair must not clobber the stored validators. + svc.store_validators(url, None, None).await; + assert_eq!( + svc.cached_validators(url).await, + (Some("\"e1\"".to_string()), None) + ); +} + // ── Test helpers ───────────────────────────────────────────────────────── -async fn make_subscription(svc: &FeedSchedulerService) -> Vec { +async fn make_subscription(svc: &FeedSchedulerService, feed_url: &str) -> Vec { let feed_id = FeedId::new(); let id_bytes = feed_id.as_bytes().to_vec(); let sub = podcast::PodcastSubscription { id: id_bytes.clone(), - feed_url: "https://example.com/podcast.xml".to_string(), + feed_url: feed_url.to_string(), title: Some("Test Podcast".to_string()), description: None, author: None, @@ -191,13 +445,13 @@ async fn make_subscription(svc: &FeedSchedulerService) -> Vec { id_bytes } -async fn make_news_feed(svc: &FeedSchedulerService) -> Vec { +async fn make_news_feed(svc: &FeedSchedulerService, url: &str) -> Vec { let feed_id = FeedId::new(); let id_bytes = feed_id.as_bytes().to_vec(); let feed = news::NewsFeed { id: id_bytes.clone(), title: "Test News".to_string(), - url: "https://example.com/news.xml".to_string(), + url: url.to_string(), site_url: None, description: None, category: None, diff --git a/crates/komide/src/test_support.rs b/crates/komide/src/test_support.rs new file mode 100644 index 00000000..85c17c3c --- /dev/null +++ b/crates/komide/src/test_support.rs @@ -0,0 +1,80 @@ +//! Shared test fixtures — scripted in-process HTTP servers for feed fetch tests. +#![cfg(test)] + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; + +/// Builds a raw HTTP/1.1 response with a correct `Content-Length` header and +/// `connection: close`. +pub(crate) fn http_response( + status: u16, + reason: &str, + extra_headers: &[(&str, &str)], + body: &[u8], +) -> Vec { + let mut head = format!("HTTP/1.1 {status} {reason}\r\n"); + for (name, value) in extra_headers { + head.push_str(&format!("{name}: {value}\r\n")); + } + head.push_str(&format!( + "content-length: {}\r\nconnection: close\r\n\r\n", + body.len() + )); + let mut bytes = head.into_bytes(); + bytes.extend_from_slice(body); + bytes +} + +/// Builds a raw HTTP/1.1 response WITHOUT `Content-Length`; the body is +/// delimited by connection close, exercising the streamed read path. +pub(crate) fn http_response_close_delimited(status: u16, reason: &str, body: &[u8]) -> Vec { + let mut bytes = format!("HTTP/1.1 {status} {reason}\r\nconnection: close\r\n\r\n").into_bytes(); + bytes.extend_from_slice(body); + bytes +} + +/// Spawns a TCP server that answers one HTTP request per scripted response, +/// in sequence, then resolves to the raw request heads it received. +/// +/// Every scripted response should carry `connection: close` so the client +/// opens a fresh connection for the next request. Write-side failures are +/// tolerated: a client that aborts mid-body (e.g. an over-cap reject) must +/// not panic the server task the test still joins. +pub(crate) async fn spawn_scripted_http( + responses: Vec>, +) -> (String, JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + + let handle = tokio::spawn(async move { + let mut requests = Vec::new(); + for response in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + + loop { + let n = stream.read(&mut chunk).await.unwrap(); + assert!(n > 0, "client closed before sending full headers"); + buf.extend_from_slice(&chunk[..n]); + if find_subslice(&buf, b"\r\n\r\n").is_some() { + break; + } + } + + if stream.write_all(&response).await.is_ok() { + stream.flush().await.ok(); + stream.shutdown().await.ok(); + } + requests.push(String::from_utf8_lossy(&buf).into_owned()); + } + requests + }); + + (base_url, handle) +} + +fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +}