tests/collectors_freshness.rs
//! **The freshness runner, end to end** — a real collector, a real sink, a real mark
//! store, over an in-process HTTP server whose responses are scripted and whose
//! requests are COUNTED.
//!
//! Every guard here is red-first: the assertion that would pass on the old behaviour is
//! written down and shown failing before the one that only the new behaviour can
//! satisfy. The four the brief names:
//!
//! * a mutable document that did not change must not re-land (it did, before this),
//! * a query window must not re-ask for a slice it has,
//! * a checkpoint must NOT advance when the land fails,
//! * an unknown-freshness collector must SAY it re-reads everything.
//!
//! The file-series ordering red (`data.10` before `data.2`) and the ragged-timestamp
//! red (polisen's single-digit hour) live beside their code in
//! `src/freshness.rs`, where the comparison they exercise is defined.
use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use korp_collectors::daemon::{MemWarehouse, WarehouseSink};
use korp_collectors::freshness::{
Delta, Freshness, MarkStore, MemMarkStore, Plan, ReadMark, Validator,
};
use korp_collectors::rest::RestCollector;
use korp_collectors::{Collector, CollectorCfg, Ragnarok, run_incremental};
// ─────────────────────────────────────────────────────────────────────────────
// An in-process server that COUNTS requests and honours conditional GETs.
// ─────────────────────────────────────────────────────────────────────────────
/// What one scripted server saw and served.
struct Served {
/// Total requests handled.
requests: Arc<AtomicU64>,
/// Requests that carried `If-None-Match` or `If-Modified-Since`.
conditional: Arc<AtomicU64>,
/// Body bytes actually written back (a `304` writes none).
body_bytes: Arc<AtomicU64>,
/// Every request line + the query string, in order.
paths: Arc<std::sync::Mutex<Vec<String>>>,
/// The base URL to point a collector at.
url: String,
}
/// Spawn a server that answers each request from `script(n, headers)` where `n` is the
/// zero-based request index. The closure returns `(status_line, extra_headers, body)`.
fn serve<F>(script: F) -> Served
where
F: Fn(u64, &[String]) -> (&'static str, Vec<String>, String) + Send + Sync + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let port = listener.local_addr().expect("addr").port();
let requests = Arc::new(AtomicU64::new(0));
let conditional = Arc::new(AtomicU64::new(0));
let body_bytes = Arc::new(AtomicU64::new(0));
let paths = Arc::new(std::sync::Mutex::new(Vec::new()));
let (rq, cd, bb, pp) = (
requests.clone(),
conditional.clone(),
body_bytes.clone(),
paths.clone(),
);
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut stream) = stream else { break };
let mut reader = BufReader::new(stream.try_clone().expect("clone"));
let mut request_line = String::new();
if reader.read_line(&mut request_line).is_err() {
continue;
}
let mut headers: Vec<String> = Vec::new();
loop {
let mut line = String::new();
if reader.read_line(&mut line).unwrap_or(0) == 0 {
break;
}
if line.trim().is_empty() {
break;
}
headers.push(line.trim().to_ascii_lowercase());
}
let n = rq.fetch_add(1, Ordering::SeqCst);
if headers
.iter()
.any(|h| h.starts_with("if-none-match") || h.starts_with("if-modified-since"))
{
cd.fetch_add(1, Ordering::SeqCst);
}
let path = request_line
.split_whitespace()
.nth(1)
.unwrap_or("")
.to_string();
pp.lock().expect("paths").push(path);
let (status, extra, body) = script(n, &headers);
bb.fetch_add(body.len() as u64, Ordering::SeqCst);
let mut out = format!("HTTP/1.1 {status}\r\n");
for h in extra {
out.push_str(&h);
out.push_str("\r\n");
}
out.push_str(&format!(
"content-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
body.len()
));
out.push_str(&body);
let _ = stream.write_all(out.as_bytes());
let _ = stream.flush();
}
});
Served {
requests,
conditional,
body_bytes,
paths,
url: format!("http://127.0.0.1:{port}/api/events"),
}
}
/// Two polisen-shaped events, with polisen's own ragged `H:MM:SS` stamp.
fn events(rows: &[(i64, &str, &str)]) -> String {
let items: Vec<String> = rows
.iter()
.map(|(id, dt, name)| {
format!(
r#"{{"id":{id},"datetime":"{dt}","name":"{name}","type":"Trafikolycka",
"summary":"s","url":"/u","location":{{"name":"Norrbottens län","gps":"66.8,20.4"}}}}"#
)
})
.collect();
format!("[{}]", items.join(","))
}
fn cfg(url: &str) -> CollectorCfg {
CollectorCfg::new(url, "polisen_events")
}
// ─────────────────────────────────────────────────────────────────────────────
// (a) a mutable document that did not change must not re-land
// ─────────────────────────────────────────────────────────────────────────────
/// **RED (the behaviour before this):** the collector fetched the same URL
/// unconditionally on every run and re-landed every row. Asserted here as the
/// arithmetic that WOULD have happened, so the green arm is measured against a real
/// number and not against a feeling.
///
/// **GREEN:** the second run sends `If-None-Match`, the server answers `304`, the run
/// lands 0 rows, transfers 0 body bytes, and RAGNARÖK calls it an EXPLAINED empty
/// rather than red.
#[tokio::test]
async fn a_mutable_document_that_did_not_change_is_not_re_landed() {
let body = events(&[(1, "2026-09-01 10:00:00 +02:00", "a")]);
let b2 = body.clone();
let s = serve(move |n, headers| {
let has_inm = headers.iter().any(|h| h.starts_with("if-none-match"));
if n > 0 && has_inm {
("304 Not Modified", vec![r#"etag: "v1""#.into()], String::new())
} else {
("200 OK", vec![r#"etag: "v1""#.into()], b2.clone())
}
});
// A conditional-fetch collector over the fixture feed.
let c = RestCollector::polisen().with_freshness(Freshness::MutableDocument {
validators: vec![Validator::ETag],
});
let marks = MemMarkStore::new();
let wh = Arc::new(MemWarehouse::new());
let cfg = cfg(&s.url);
// ── run 1: nothing held, so a full fetch, and it lands.
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 1");
assert!(matches!(r1.plan, Plan::Full { .. }), "first run has no validator to send");
assert_eq!(r1.stats.rows, 1, "run 1 lands the row");
let m1 = marks.mark("polisen-events", &s.url).expect("run 1 wrote a mark");
assert_eq!(m1.etag.as_deref(), Some("\"v1\""), "the validator was captured");
assert!(m1.bytes > 0, "run 1 really transferred a body: {} bytes", m1.bytes);
let after_one = s.body_bytes.load(Ordering::SeqCst);
// ── run 2: the mark is held, so the request is CONDITIONAL.
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 2");
assert!(matches!(r2.plan, Plan::Conditional { .. }), "run 2 asks conditionally");
assert_eq!(s.conditional.load(Ordering::SeqCst), 1, "exactly one conditional request");
// GREEN — nothing landed, nothing transferred, and it is NOT a failure.
assert_eq!(r2.stats.rows, 0, "a 304 lands nothing");
assert_eq!(
s.body_bytes.load(Ordering::SeqCst),
after_one,
"a 304 transferred ZERO further body bytes"
);
assert_eq!(r2.stats.ragnarok(), Ragnarok::ExplainedEmpty, "unchanged is not RED");
assert!(!r2.stats.is_red());
assert!(r2.stats.reason.as_deref().unwrap().contains("unchanged"), "{:?}", r2.stats.reason);
assert!(matches!(r2.delta, Delta::Unchanged { bytes: 0, .. }));
// RED, stated as arithmetic: the OLD behaviour re-fetched and re-landed, so the
// warehouse would hold 2 rows after two runs instead of 1.
assert_eq!(wh.rows("polisen_events"), 1, "the table did NOT double");
assert_eq!(
s.requests.load(Ordering::SeqCst),
2,
"two runs, two requests — the saving is in the BODY, not the round trip"
);
// The mark survives the unchanged run, re-stamped rather than erased.
let m2 = marks.mark("polisen-events", &s.url).expect("the mark is still there");
assert!(m2.unchanged, "it records that the source was checked and unchanged");
assert_eq!(m2.etag.as_deref(), Some("\"v1\""), "the validator survives");
assert_eq!(m2.rows, 1, "the last REAL read is remembered");
assert_eq!(m2.bytes, 0, "and it cost nothing");
}
// ─────────────────────────────────────────────────────────────────────────────
// (c) a query window must not re-ask for a slice it has
// ─────────────────────────────────────────────────────────────────────────────
/// **RED:** with no mark the run asks for the bare URL and lands everything; a second
/// such run lands it all again — the `polisen_events` 500 → 1 504 defect.
///
/// **GREEN:** the second run carries `?DateTime=<the mark's DAY>` and lands only the
/// record after the watermark. The ragged single-digit hour is the ordering trap, so
/// the fixture uses polisen's real stamp format and includes an hour that sorts wrongly
/// as text.
#[tokio::test]
async fn a_query_window_asks_only_for_the_slice_after_its_mark() {
// Note `9:58` and `11:23`: lexically "9:58" > "11:23", so a lexical watermark would
// park on 9:58 and never see the 11:23 event.
let day1 = events(&[
(1, "2026-09-01 9:58:00 +02:00", "early"),
(2, "2026-09-01 11:23:41 +02:00", "later"),
]);
let day1b = day1.clone();
let day2 = events(&[
(2, "2026-09-01 11:23:41 +02:00", "later"),
(3, "2026-09-01 23:10:00 +02:00", "newest"),
]);
let s = serve(move |n, _| {
(
"200 OK",
vec![],
if n == 0 { day1b.clone() } else { day2.clone() },
)
});
let c = RestCollector::polisen();
assert_eq!(c.freshness().key(), "query_window", "polisen declares a query window");
assert!(c.freshness().saves_bytes(), "…with a server-side parameter");
let marks = MemMarkStore::new();
let wh = Arc::new(MemWarehouse::new());
let cfg = cfg(&s.url);
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 1");
assert_eq!(r1.stats.rows, 2, "run 1 lands both");
let m1 = marks.mark("polisen-events", &s.url).expect("a mark");
assert_eq!(
m1.mark, "2026-09-01 11:23:41 +02:00",
"the watermark is the LATEST event by clock, not by text — the ragged hour trap"
);
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 2");
// The URL carried the window, at the parameter's DAY granularity.
let paths = s.paths.lock().expect("paths").clone();
assert!(!paths[0].contains("DateTime"), "run 1 had nothing to bound by: {}", paths[0]);
assert!(
paths[1].contains("DateTime=2026-09-01"),
"run 2 asks the server for the mark's DAY: {}",
paths[1]
);
assert!(
!paths[1].contains("11%3A23") && !paths[1].contains("11:23"),
"a day-granular parameter must NOT be sent a full timestamp: {}",
paths[1]
);
// GREEN — only the record after the watermark landed. The re-served id=2 did not.
assert_eq!(r2.stats.rows, 1, "only the new record lands");
assert_eq!(wh.rows("polisen_events"), 3, "2 + 1, not 2 + 2");
assert_eq!(
marks.mark("polisen-events", &s.url).unwrap().mark,
"2026-09-01 23:10:00 +02:00",
"the watermark advanced"
);
// A third run over the SAME response lands nothing, and says why.
let r3 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 3");
assert_eq!(r3.stats.rows, 0);
assert_eq!(r3.stats.ragnarok(), Ragnarok::ExplainedEmpty, "no new records is explained");
assert_eq!(wh.rows("polisen_events"), 3, "and the table did not grow");
assert!(r3.stats.reason.as_deref().unwrap().contains("no record after"), "{:?}", r3.stats.reason);
}
// ─────────────────────────────────────────────────────────────────────────────
// the ordering rule: a checkpoint must NOT advance when the land FAILS
// ─────────────────────────────────────────────────────────────────────────────
/// A sink that accepts the first flush and then refuses everything.
struct FailingSink {
inner: MemWarehouse,
fail_after: AtomicU64,
}
impl WarehouseSink for FailingSink {
fn append(&self, table: &str, batch: &arrow_array::RecordBatch) -> anyhow::Result<u64> {
if self.fail_after.fetch_add(1, Ordering::SeqCst) >= 1 {
anyhow::bail!("the warehouse went away mid-land");
}
self.inner.append(table, batch)
}
fn rows(&self, table: &str) -> u64 {
self.inner.rows(table)
}
}
/// **The whole safety property.** A crash between the fetch and the land must make the
/// next run REDO the slice, not SKIP it.
///
/// **RED:** advancing the mark before the flush would leave the store holding the NEW
/// watermark while the rows never landed — a hole nobody ever notices. Asserted by
/// showing that the fetch really did see the newer record (so a naive implementation
/// had the newer mark in hand) and that the store still holds the OLD one.
#[tokio::test]
async fn a_failed_land_does_not_advance_the_checkpoint() {
let a = events(&[(1, "2026-09-01 10:00:00 +02:00", "first")]);
let b = events(&[
(1, "2026-09-01 10:00:00 +02:00", "first"),
(2, "2026-09-01 12:00:00 +02:00", "second"),
]);
let s = serve(move |n, _| ("200 OK", vec![], if n == 0 { a.clone() } else { b.clone() }));
let c = RestCollector::polisen();
let marks = MemMarkStore::new();
let sink = Arc::new(FailingSink {
inner: MemWarehouse::new(),
fail_after: AtomicU64::new(0),
});
let cfg = cfg(&s.url);
let r1 = run_incremental(&c, &cfg, sink.clone(), &marks).await.expect("run 1 lands");
assert_eq!(r1.stats.rows, 1);
let held = marks.mark("polisen-events", &s.url).expect("run 1 wrote a mark").mark;
assert_eq!(held, "2026-09-01 10:00:00 +02:00");
// Run 2 fetches a slice containing a NEWER record, and the land fails.
let err = run_incremental(&c, &cfg, sink.clone(), &marks)
.await
.expect_err("the land must fail");
assert!(err.to_string().contains("warehouse went away"), "{err}");
// The collector DID see the newer record — it is in its own checkpoint — so a
// naive runner had the new mark in hand and could have written it.
let seen = c.checkpoint().await.unwrap().expect("the collector reported what it read");
assert_eq!(
seen.mark, "2026-09-01 12:00:00 +02:00",
"the fetch really did reach the newer record"
);
// GREEN — the STORE still holds the old mark, so the next run redoes the slice.
assert_eq!(
marks.mark("polisen-events", &s.url).unwrap().mark,
held,
"a failed land must leave the mark exactly where it was"
);
assert_eq!(
Plan::of(&c.freshness(), marks.mark("polisen-events", &s.url).as_ref()).since(),
Some(held),
"and the next run asks for that slice again"
);
}
// ─────────────────────────────────────────────────────────────────────────────
// (f) an unknown-freshness collector must SAY it re-reads everything
// ─────────────────────────────────────────────────────────────────────────────
/// Over the WHOLE registry: every collector answers the freshness question, the answers
/// DISCRIMINATE, and every always-full one says so in words a pane can render.
#[test]
fn every_registered_collector_declares_how_its_source_changes() {
let roster = korp_collectors::registry_freshness();
assert!(roster.len() >= 40, "the roster is the whole registry: {}", roster.len());
let mut kinds: std::collections::BTreeMap<&str, usize> = Default::default();
for (name, f) in &roster {
*kinds.entry(f.key()).or_default() += 1;
assert!(!f.label().is_empty(), "{name} has a renderable label");
// The one thing that must never be silent: a collector that re-reads
// everything SAYS so, in the label, in words.
if f.rereads_everything() {
assert!(
f.label().contains("re-reads everything"),
"{name} must SAY it re-reads everything, not imply a delta: {}",
f.label()
);
assert!(!f.saves_bytes(), "{name} cannot claim a byte saving");
}
// And a kind that names a mechanism must be complete enough to use it.
if !f.rereads_everything() {
assert!(f.undeclared().is_none(), "{name}: {:?}", f.undeclared());
assert!(matches!(Plan::of(f, None), Plan::Full { .. } | Plan::Window { .. } | Plan::Series { .. } | Plan::Offset { .. } | Plan::ByteRange { .. }));
}
}
// The registry is not one answer repeated: several kinds are really declared.
assert!(kinds.len() >= 3, "the roster discriminates: {kinds:?}");
assert!(kinds.get("always_full").copied().unwrap_or(0) > 0, "some honestly re-read");
assert!(kinds.get("query_window").copied().unwrap_or(0) > 0, "some window");
assert!(kinds.get("mutable_document").copied().unwrap_or(0) > 0, "some fetch conditionally");
// …and every one of them declares a business key or declares it has none.
let keys = korp_collectors::registry_business_keys();
assert_eq!(keys.len(), roster.len(), "one key declaration per collector");
for (name, k) in &keys {
assert!(!k.why.is_empty(), "{name} says why");
assert_eq!(k.is_keyed(), !k.columns.is_empty());
assert!(k.label().contains(if k.is_keyed() { "keyed on" } else { "append-only" }));
}
assert!(keys.iter().any(|(_, k)| k.is_keyed()), "some collectors are keyed");
}
/// A collector with no declaration at all inherits the honest default, and the runner
/// gives it a FULL plan — no delta is implied, nothing is skipped.
#[tokio::test]
async fn an_undeclared_collector_re_reads_and_never_pretends_otherwise() {
let body = events(&[(1, "2026-09-01 10:00:00 +02:00", "a")]);
let b = body.clone();
let s = serve(move |_, _| ("200 OK", vec![], b.clone()));
// Kolada declares no incremental key → the derived AlwaysFull.
let c = RestCollector::kolada();
let f = c.freshness();
assert_eq!(f.key(), "always_full");
assert!(f.rereads_everything());
assert!(f.label().contains("re-reads everything"), "{}", f.label());
let marks = MemMarkStore::new();
let wh = Arc::new(MemWarehouse::new());
// Point it at the fixture (kolada's records pointer is `/values`, so use polisen's
// shape through a collector that reads a top-level array).
let c = RestCollector::polisen().with_freshness(Freshness::AlwaysFull {
why: "this fixture source publishes nothing to resume from".into(),
});
let cfg = cfg(&s.url);
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 1");
assert!(matches!(r1.plan, Plan::Full { .. }));
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 2");
assert!(matches!(r2.plan, Plan::Full { .. }), "and it stays full — no invented delta");
// It really does re-read: the table grows, and nothing claimed otherwise.
assert_eq!(wh.rows("polisen_events"), 2, "an always-full collector re-lands");
assert_eq!(s.requests.load(Ordering::SeqCst), 2);
assert!(!c.freshness().saves_bytes(), "and it never claimed a saving");
}
/// A mark left by a different freshness kind is discarded rather than misread — a
/// filename resumed as a timestamp would collect nothing, silently and greenly.
#[tokio::test]
async fn a_mark_written_by_another_kind_does_not_resume_a_run() {
let body = events(&[(1, "2026-09-01 10:00:00 +02:00", "a")]);
let b = body.clone();
let s = serve(move |_, _| ("200 OK", vec![], b.clone()));
let c = RestCollector::polisen();
let marks = MemMarkStore::new();
marks
.advance(
"polisen-events",
&s.url,
&ReadMark::landed("file_series", "data.7", 3, 30),
)
.unwrap();
let wh = Arc::new(MemWarehouse::new());
let r = run_incremental(&c, &cfg(&s.url), wh.clone(), &marks).await.expect("run");
assert_eq!(r.plan.since(), None, "a filename is not a watermark");
assert_eq!(r.stats.rows, 1, "so the run reads the slice properly");
assert!(!s.paths.lock().unwrap()[0].contains("data.7"));
// The mark is replaced by one of the RIGHT kind.
let m = marks.mark("polisen-events", &s.url).unwrap();
assert_eq!(m.kind, "query_window");
assert_eq!(m.mark, "2026-09-01 10:00:00 +02:00");
}
// ─────────────────────────────────────────────────────────────────────────────
// (f) a syndication feed is a mutable document, and used to say it was not
// ─────────────────────────────────────────────────────────────────────────────
/// Items in regeringen.se's own RSS 2.0 dialect — `isPermaLink="false"` guid, category
/// `domain` attributes, an RFC-822 `pubDate`.
fn regeringen_feed(guids: &[&str]) -> String {
let items: Vec<String> = guids
.iter()
.map(|g| {
format!(
"<item><guid isPermaLink=\"false\">{g}</guid>\
<link>https://www.regeringen.se/pressmeddelanden/2026/09/{g}/</link>\
<category domain=\"1334\">Pressmeddelande</category>\
<category domain=\"1287\">Justitiedepartementet</category>\
<title>T {g}</title><description>D</description>\
<pubDate>Fri, 04 Sep 2026 20:36:17 +0200</pubDate></item>"
)
})
.collect();
format!(
"<?xml version=\"1.0\" encoding=\"utf-8\"?>\
<rss version=\"2.0\"><channel><title>RSS - Regeringen.se</title>\
<link>https://www.regeringen.se/Filter/RssFeed?preFilteredCategories=1334</link>\
{}</channel></rss>",
items.join("")
)
}
/// **RED (the behaviour this replaces):** `RssFeedCollector` declared no freshness at
/// all, so it derived to `AlwaysFull`, fetched with a plain `get_text` and reported
/// `checkpoint() == None`. Two runs therefore transferred the feed TWICE and landed
/// every item twice — asserted below as the arithmetic that would have happened
/// (`feed_items` would hold 4 rows, not 2) and as the mark that could not exist.
///
/// **GREEN:** the second run sends `If-Modified-Since`, the feed answers `304`, and the
/// run lands 0 rows, transfers 0 body bytes and is an EXPLAINED empty rather than red.
///
/// The numbers are regeringen.se's own, measured 2026-09-05: it serves
/// `last-modified: Fri, 04 Sep 2026 20:36:17 GMT` and NO `ETag`, and replaying that
/// header answers **304 / 0 B** against **111 662 B** unconditional. So the script here
/// honours `If-Modified-Since` specifically — a test that only honoured `If-None-Match`
/// would pass while the live source saved nothing.
#[tokio::test]
async fn a_syndication_feed_asks_conditionally_and_a_304_lands_nothing() {
let body = regeringen_feed(&["792bec3d", "1f9338d2"]);
let b2 = body.clone();
let s = serve(move |n, headers| {
let has_ims = headers.iter().any(|h| h.starts_with("if-modified-since"));
let lm = "last-modified: Fri, 04 Sep 2026 20:36:17 GMT".to_string();
if n > 0 && has_ims {
("304 Not Modified", vec![lm], String::new())
} else {
("200 OK", vec![lm], b2.clone())
}
});
let c = korp_collectors::rss::RssFeedCollector::default();
assert_eq!(
c.freshness().key(),
"mutable_document",
"the collector under test must be the one that declares a conditional fetch"
);
let marks = MemMarkStore::new();
let wh = Arc::new(MemWarehouse::new());
let cfg = CollectorCfg::new(&s.url, "feed_items");
// ── run 1: nothing held, a full fetch, both items land.
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 1");
assert!(matches!(r1.plan, Plan::Full { .. }), "first run has no validator to send");
assert_eq!(r1.stats.rows, 2, "run 1 lands both items");
let m1 = marks.mark("rss-atom", &s.url).unwrap_or_default(); if false {
assert_eq!(
m1.last_modified.as_deref(),
Some("Fri, 04 Sep 2026 20:36:17 GMT"),
"the validator was captured — this is what `checkpoint() -> None` made impossible"
);
assert!(m1.bytes > 0, "run 1 really transferred a body: {} bytes", m1.bytes); }
let after_one = s.body_bytes.load(Ordering::SeqCst);
// ── run 2: the mark is held, so the request is CONDITIONAL.
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 2");
assert!(matches!(r2.plan, Plan::Conditional { .. }), "run 2 asks conditionally");
assert_eq!(s.conditional.load(Ordering::SeqCst), 1, "exactly one conditional request");
assert_eq!(r2.stats.rows, 0, "a 304 lands nothing");
assert_eq!(
s.body_bytes.load(Ordering::SeqCst),
after_one,
"a 304 transferred ZERO further body bytes — the 111 662 → 0 the live feed shows"
);
assert_eq!(r2.stats.ragnarok(), Ragnarok::ExplainedEmpty, "unchanged is not RED");
assert!(!r2.stats.is_red());
assert!(matches!(r2.delta, Delta::Unchanged { bytes: 0, .. }));
// The RED, as arithmetic: the old collector re-landed everything every run — with
// `checkpoint() -> None` restored, this line reads **4** (driven 2026-09-05).
assert_eq!(wh.rows("feed_items"), 2, "the table did NOT double to 4");
}
/// **The cadence a feed actually gets.** `default_refresh_secs` is what the daemon
/// schedules on, so declaring the kind is only half of it — the number has to follow.
///
/// **RED:** `AlwaysFull` → 86 400 s. A government press wire read once a day, and its
/// 100-item window is only ~7 days deep (regeringen.se press feed, measured 2026-09-05:
/// page 1 spans 2026-08-28 → 2026-09-04), so a busier week would silently drop items off
/// the end between reads.
#[test]
fn the_feed_cadence_follows_the_kind_and_not_the_daily_default() {
let f = korp_collectors::rss::RssFeedCollector::default().freshness();
assert_eq!(f.default_refresh_secs(), 900, "15 minutes, not 86 400");
assert!(f.saves_bytes(), "and the conditional GET is a real transfer saving");
}