src/freshness.rs
//! **How a source CHANGES** — the declaration, the mark a run leaves behind, and
//! the plan the next run derives from the two.
//!
//! # The hole this closes
//!
//! [`Bounds`](crate::Bounds) and [`Checkpoint`](crate::Checkpoint) declared the CDC
//! seam from the first commit — "which slice to collect", "where a CDC-aware collector
//! got to". **Nothing filled it.** `since` was set in exactly two places and both were
//! tests; no checkpoint survived a process; every collector re-read its whole source on
//! every run. `polisen_events` went 500 → 1 504 rows by re-fetching and re-landing the
//! same 500 events three times, not by a delta.
//!
//! The missing fact is one only the COLLECTOR knows: **how does this particular source
//! change?** A file that is rewritten in place, a directory that gains `data.1`,
//! `data.2`, an API with a `since=` parameter and a log with an offset are four
//! different questions, and no generic runner can guess which one it is looking at.
//! So the collector declares it, exactly as it declares
//! [`requirements()`](crate::Collector::requirements).
//!
//! # The six kinds
//!
//! | kind | how it changes | what the mark is | what a re-run saves |
//! |---|---|---|---|
//! | [`MutableDocument`](Freshness::MutableDocument) | same URL, rewritten in place | the `ETag` / `Last-Modified` validator | the whole body, on a `304` |
//! | [`FileSeries`](Freshness::FileSeries) | new names appear (`data.1`, `data.2`, dated) | the last NAME consumed | every file already consumed |
//! | [`QueryWindow`](Freshness::QueryWindow) | an API grows, and takes a lower bound | the last timestamp/id seen | rows always; bytes when the server takes the parameter |
//! | [`AppendOnlyStream`](Freshness::AppendOnlyStream) | a log only ever grows | the byte/record OFFSET | everything before the offset |
//! | [`ByteRangeColumnar`](Freshness::ByteRangeColumnar) | a big columnar object is re-cut | the object's version token | every row-group column chunk already held |
//! | [`AlwaysFull`](Freshness::AlwaysFull) | **unknown — it re-reads everything** | nothing | nothing, and it SAYS so |
//!
//! [`AlwaysFull`] is not a gap in the taxonomy, it is the honest sixth answer. A
//! collector that cannot delta must say so, because the alternative — a UI that shows
//! "incremental" over a collector re-landing its whole source — is the same class of
//! lie as an unexplained empty. [`Freshness::rereads_everything`] is the one call a
//! pane makes to render "this one re-reads everything".
//!
//! # An unchanged source is not a failure
//!
//! RAGNARÖK says a 0-row collect is RED unless the collector explains it. A `304 Not
//! Modified` **is** an explanation, and a good one: the source provably did not change,
//! so landing nothing is the correct outcome. [`unchanged_reason`] builds the sentence,
//! and it deliberately does not use [`crate::RAGNAROK_UNEXPLAINED`] — it reads back
//! through [`CollectStats::ragnarok`](crate::CollectStats::ragnarok) as
//! [`Ragnarok::ExplainedEmpty`](crate::Ragnarok::ExplainedEmpty), never as `Red`.
//!
//! # The mark advances only AFTER a successful land
//!
//! [`MarkStore::advance`] is called by the runner after the sink has flushed, never
//! before. A crash between the fetch and the land must make the next run REDO the
//! slice, not SKIP it: re-landing rows is a duplicate an idempotent sink or a later
//! compaction can absorb, while a skipped slice is a hole nobody will ever notice.
//! [`Plan`] carries no ability to write; only the runner, holding a landed row count,
//! can advance a mark. That is the ordering rule made structural rather than
//! remembered.
//!
//! # The state is DERIVED, never typed
//!
//! Same discipline as [`crate::requires`]: nothing stores "is incremental". What is
//! persisted is one [`ReadMark`] per `(collector, source)` — the facts of the last
//! successful run — and every word the pane shows is computed from it plus the
//! declaration. A stored "incremental: true" flag would go stale the moment a source
//! stopped sending `ETag`s.
use std::collections::BTreeMap;
use std::sync::Mutex;
use serde::{Deserialize, Serialize};
// ─────────────────────────────────────────────────────────────────────────────
// Ordering — the one place "is this mark newer than that one" is answered
// ─────────────────────────────────────────────────────────────────────────────
/// **How two marks compare.** The single writer of "is this newer", shared by
/// [`FileSeries`](Freshness::FileSeries) (is this filename after the one we consumed?)
/// and [`QueryWindow`](Freshness::QueryWindow) (is this record after the watermark?).
///
/// It exists because *lexical* is wrong for two of the three real cases and the
/// wrongness is silent:
///
/// * `data.10` sorts BEFORE `data.2` as text, so a lexical file series re-consumes
/// every file from `data.2` onward on every run, forever.
/// * polisen stamps `2026-09-03 1:23:41 +02:00` — a **single-digit hour**. Lexically
/// `9:00` > `11:00`, so a lexical watermark parks itself at the last single-digit
/// hour of the day and drops everything after 09:59 until the next midnight.
///
/// Both were found by running the real sources, not by reading them.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MarkOrder {
/// Compare as text. Correct for fixed-width ISO-8601 (`2026-09-03T01:23:41Z`),
/// for zero-padded names, and for opaque ids that only need equality.
#[default]
Lexical,
/// Compare by the trailing run of digits (`data.1` < `data.2` < `data.10`).
/// Names with no trailing digits fall back to a lexical compare of the whole name.
Numeric,
/// Compare as a calendar timestamp, after normalising every numeric component to
/// a fixed width. For sources whose text form is not lexically sortable.
Timestamp,
}
impl MarkOrder {
/// The stable machine word.
pub fn key(self) -> &'static str {
match self {
MarkOrder::Lexical => "lexical",
MarkOrder::Numeric => "numeric",
MarkOrder::Timestamp => "timestamp",
}
}
/// **The comparable form of a mark** — compare these, never the raw strings.
///
/// Deliberately a pure `String → String` function: it is testable on its own, it
/// has no clock and no locale, and both call sites (the file series and the record
/// watermark) go through it, so they cannot disagree about what "newer" means.
pub fn sort_key(self, raw: &str) -> String {
match self {
MarkOrder::Lexical => raw.to_string(),
MarkOrder::Numeric => {
// The trailing digit run, zero-padded so 2 < 10. Everything before it
// stays as a prefix, so `a.9` and `b.1` still order by their stem.
let bytes = raw.as_bytes();
let mut i = bytes.len();
while i > 0 && bytes[i - 1].is_ascii_digit() {
i -= 1;
}
if i == bytes.len() {
return raw.to_string();
}
format!("{}{:0>20}", &raw[..i], &raw[i..])
}
MarkOrder::Timestamp => {
// Zero-pad every digit run to 4 so `1:23` and `11:23` compare by value.
// The separators are kept in place, so the year/month/day/hour ordering
// of an ISO-ish stamp survives, and a stamp that is ALREADY fixed-width
// is unchanged in relative order.
let mut out = String::with_capacity(raw.len() + 16);
let mut digits = String::new();
for ch in raw.chars() {
if ch.is_ascii_digit() {
digits.push(ch);
} else {
if !digits.is_empty() {
out.push_str(&format!("{digits:0>4}"));
digits.clear();
}
out.push(ch);
}
}
if !digits.is_empty() {
out.push_str(&format!("{digits:0>4}"));
}
out
}
}
}
/// Is `candidate` strictly after `mark`? A `None` mark means "we have nothing yet",
/// and everything is after nothing.
pub fn is_after(self, candidate: &str, mark: Option<&str>) -> bool {
match mark {
None => true,
Some(m) => self.sort_key(candidate) > self.sort_key(m),
}
}
/// The newest of `candidates` by this order — the high-watermark a run reports.
pub fn newest<'a>(self, candidates: impl IntoIterator<Item = &'a str>) -> Option<String> {
candidates
.into_iter()
.max_by(|a, b| self.sort_key(a).cmp(&self.sort_key(b)))
.map(str::to_string)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The window a server-side `since` parameter can actually express
// ─────────────────────────────────────────────────────────────────────────────
/// **How coarse the server's lower-bound parameter is.**
///
/// This is not decoration. polisen's `?DateTime=` takes a DAY (`2026-09-02`), so asking
/// it for the watermark `2026-09-02 14:31:07` verbatim returns nothing at all; asking
/// for `2026-09-02` returns that whole day, from which the run keeps the records after
/// the watermark. A collector that sent the exact mark to a day-granular parameter would
/// go silently empty — a green run that collects nothing, forever.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WindowGranularity {
/// The parameter takes the mark verbatim.
#[default]
Exact,
/// The parameter takes a date (`YYYY-MM-DD`) — the mark is truncated to its day.
Day,
/// The parameter takes a month (`YYYY-MM`).
Month,
/// The parameter takes a year (`YYYY`).
Year,
}
impl WindowGranularity {
/// The stable machine word.
pub fn key(self) -> &'static str {
match self {
WindowGranularity::Exact => "exact",
WindowGranularity::Day => "day",
WindowGranularity::Month => "month",
WindowGranularity::Year => "year",
}
}
/// **The value to actually PUT in the query parameter** for a given mark.
///
/// Always the truncation, never the mark — and the run must therefore still filter
/// records locally, because the server hands back the whole coarse window. That is
/// the deal: the parameter buys BYTES, the local filter buys ROWS, and both are
/// needed.
pub fn param_value(self, mark: &str) -> String {
let take = match self {
WindowGranularity::Exact => return mark.to_string(),
WindowGranularity::Day => 10, // YYYY-MM-DD
WindowGranularity::Month => 7, // YYYY-MM
WindowGranularity::Year => 4, // YYYY
};
mark.chars().take(take).collect()
}
}
// ─────────────────────────────────────────────────────────────────────────────
// HTTP validators — what a mutable document offers to prove it did not change
// ─────────────────────────────────────────────────────────────────────────────
/// One conditional-request validator a source serves.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Validator {
/// `ETag` ⇄ `If-None-Match` — exact, and what a proxy or CDN usually offers.
ETag,
/// `Last-Modified` ⇄ `If-Modified-Since` — one-second resolution, and what the
/// two live keyless sources on this box actually serve (USGS and SMHI both do;
/// polisen serves NEITHER, which is why polisen is a query window and not a
/// mutable document).
LastModified,
}
impl Validator {
/// The response header the server sends.
pub fn response_header(self) -> &'static str {
match self {
Validator::ETag => "etag",
Validator::LastModified => "last-modified",
}
}
/// The request header the next run sends back.
pub fn request_header(self) -> &'static str {
match self {
Validator::ETag => "if-none-match",
Validator::LastModified => "if-modified-since",
}
}
/// The stable machine word.
pub fn key(self) -> &'static str {
match self {
Validator::ETag => "etag",
Validator::LastModified => "last_modified",
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The declaration
// ─────────────────────────────────────────────────────────────────────────────
/// **How this source changes** — declared by the collector, because only it knows.
///
/// The sibling of [`Requirements`](crate::requires::Requirements): that one answers
/// *what does it need before it can run*, this one answers *what has changed since it
/// last ran*. Both are declarations with a derived default, both are overridden per
/// collector, and neither stores state.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Freshness {
/// **(a) A mutable document.** The same URL, rewritten in place. The next run
/// sends the stored validators back and a `304` means "nothing to do" — the body
/// is never transferred and nothing is re-landed.
MutableDocument {
/// Which validators this source serves. Empty is a declaration that the
/// source offers NONE, which makes conditional fetching impossible — see
/// [`Freshness::undeclared`].
validators: Vec<Validator>,
},
/// **(b) A series of files.** `data.1`, `data.2`, dated names, a directory index
/// or an S3 listing. The collector says where the names are enumerated and how to
/// order them; the mark is the last name CONSUMED, and the next run takes only
/// the names after it.
FileSeries {
/// Where the names come from — a directory index URL, an S3 `list-objects`
/// prefix, a manifest. **Empty = undeclared**; a series with no listing
/// cannot enumerate anything.
listing: String,
/// The suffix that identifies a member of the series (`.parquet`, `.csv.gz`).
/// Empty accepts every name in the listing.
name_suffix: String,
/// How the names order. `data.10` after `data.2` needs
/// [`MarkOrder::Numeric`]; this is the field that makes that explicit
/// instead of hoped-for.
order: MarkOrder,
},
/// **(c) A query window.** An API that grows and takes a lower bound.
QueryWindow {
/// The query parameter carrying the lower bound. **`None` is a real and
/// common answer**: the server has no such parameter, so the whole body is
/// still fetched and only the ROWS are narrowed, locally. Saying `None` is
/// what keeps [`Freshness::saves_bytes`] from claiming a transfer saving that
/// does not happen.
param: Option<String>,
/// RFC-6901 pointer, relative to each record, to the field the mark is taken
/// from — the same pointer
/// [`RestCollector::with_bounds_field`](crate::rest::RestCollector::with_bounds_field)
/// already sets.
record_field: String,
/// How the field's values order.
order: MarkOrder,
/// How coarse `param` is.
granularity: WindowGranularity,
},
/// **(d) An append-only stream or log.** Only ever grows, and is addressed by an
/// offset rather than a time.
AppendOnlyStream {
/// The query parameter carrying the offset. **Empty = undeclared.**
offset_param: String,
},
/// **(e) A byte-range columnar object** (parquet on S3 or on a filesystem).
///
/// The delta mechanism is **`nornir-parquet-delta`** (edda `ac22b5e`) and this
/// crate does not contain a second one: it plans `footer + need → byte ranges`,
/// keeps a sparse local mirror, and holds an opaque version token that changes
/// when the file does. This variant DECLARES the delegation and names the object;
/// the HOST executes it, because korp-collectors is a leaf and may not take the
/// dependency (it is also on `parquet` 59 while this crate is on 58).
///
/// The version token is [`ReadMark::etag`] — `nornir_parquet_delta::Source`
/// documents its token as deliberately opaque and taken from whatever the system
/// already has, and an `ETag` is exactly that.
ByteRangeColumnar {
/// The object URL / path. **Empty = undeclared.**
object: String,
},
/// **(f) Unknown, or genuinely not incremental** — it re-reads everything, and
/// says so.
///
/// The honest sixth answer, and the derived default for a collector that has not
/// overridden [`Collector::freshness`](crate::Collector::freshness). A pane
/// renders this as "re-reads everything" rather than as a delta that is not
/// happening.
AlwaysFull {
/// Why — an operator-facing line. Never empty.
why: String,
},
}
impl Default for Freshness {
/// **A source nobody has described re-reads everything.** The default has to be
/// one of the six, and any incremental kind would be a claim about a source
/// nobody has looked at.
fn default() -> Self {
Freshness::AlwaysFull {
why: "no freshness was declared for this source".into(),
}
}
}
impl Freshness {
/// The stable machine key (the wire, `$KORP_STATE`, the CLI, tests).
pub fn key(&self) -> &'static str {
match self {
Freshness::MutableDocument { .. } => "mutable_document",
Freshness::FileSeries { .. } => "file_series",
Freshness::QueryWindow { .. } => "query_window",
Freshness::AppendOnlyStream { .. } => "append_only_stream",
Freshness::ByteRangeColumnar { .. } => "byte_range_columnar",
Freshness::AlwaysFull { .. } => "always_full",
}
}
/// **How long to wait before reading this source again, when nobody has said.**
///
/// The daemon loop used a flat 3600 for every collector in the roster. One number
/// cannot be right for six kinds of source, and it was wrong in both directions at
/// once: an incident feed read hourly is stale by eleven-twelfths of every hour,
/// and a national data catalogue re-read hourly is twenty-three wasted full fetches
/// a day, per collector, forever.
///
/// The cadence follows from the KIND, so it lives here beside the kinds rather than
/// in the loop that sleeps:
///
/// | kind | wait | why |
/// |---|---|---|
/// | `QueryWindow` | 5 min | it grows continuously and takes a lower bound, so a short wait costs one narrow query |
/// | `AppendOnlyStream` | 5 min | same shape: an offset forward, nothing re-read |
/// | `MutableDocument` | 15 min | a conditional fetch is nearly free, but it is still a round trip and a `304` is all most of them return |
/// | `FileSeries` | 15 min | a new member appears on the source's own schedule; the check is one listing |
/// | `ByteRangeColumnar` | 15 min | the check is a footer read, cheap but not free |
/// | `AlwaysFull` | 24 h | every read transfers EVERYTHING. This is the one kind where the interval is the whole cost, so it is the one kind that must be rare |
///
/// This is a DEFAULT, never a ceiling. An operator's own `refresh_secs` outranks it,
/// because they can see something the declaration cannot.
pub fn default_refresh_secs(&self) -> u64 {
match self {
// Incremental and cheap per read: the window or the offset only moves
// forward, so a short wait buys freshness for one narrow query.
Freshness::QueryWindow { .. } | Freshness::AppendOnlyStream { .. } => 300,
// Incremental but a round trip each time.
Freshness::MutableDocument { .. }
| Freshness::FileSeries { .. }
| Freshness::ByteRangeColumnar { .. } => 900,
// NOT incremental: every read is a full transfer, so the interval IS the
// cost. A source that re-reads everything must re-read it rarely.
Freshness::AlwaysFull { .. } => 86_400,
}
}
/// Every kind's key, in declaration order — what a test-matrix sweeps and what a
/// pane's legend lists.
pub const KEYS: &'static [&'static str] = &[
"mutable_document",
"file_series",
"query_window",
"append_only_stream",
"byte_range_columnar",
"always_full",
];
/// A conditional-fetch declaration over the usual pair of validators — the
/// shape almost every mutable document has.
pub fn conditional() -> Self {
Freshness::MutableDocument {
validators: vec![Validator::ETag, Validator::LastModified],
}
}
/// A query window narrowed **only locally**: the server takes no lower bound, so
/// the full body is fetched and the records are filtered on this side. Rows are a
/// true delta; bytes are not, and [`saves_bytes`](Self::saves_bytes) says so.
pub fn record_watermark(record_field: impl Into<String>, order: MarkOrder) -> Self {
Freshness::QueryWindow {
param: None,
record_field: record_field.into(),
order,
granularity: WindowGranularity::Exact,
}
}
/// A query window the server narrows too.
pub fn query_window(
param: impl Into<String>,
record_field: impl Into<String>,
order: MarkOrder,
granularity: WindowGranularity,
) -> Self {
Freshness::QueryWindow {
param: Some(param.into()),
record_field: record_field.into(),
order,
granularity,
}
}
/// **The default derived from what every collector already publishes** — its
/// declared per-record incremental key, if it has one.
///
/// The mirror of
/// [`Requirements::derived`](crate::requires::Requirements::derived), and for the
/// same reason: a collector that overrides nothing still ANSWERS, and answers from
/// its own words. A collector with a declared record key gets a local-only
/// watermark, which is safe (it never asks the server for anything new) and true
/// (it really does drop rows it already has). A collector with no key gets
/// [`AlwaysFull`](Freshness::AlwaysFull) — never a delta nobody checked.
///
/// The order defaults to [`MarkOrder::Timestamp`], not `Lexical`: a derived key is
/// a datetime field, and `Timestamp` is a superset of `Lexical` for the
/// fixed-width ISO stamps where lexical was already right, while being the only
/// correct answer for the ragged ones.
pub fn derived(record_key: Option<&str>) -> Self {
match record_key {
Some(k) if !k.trim().is_empty() => {
Freshness::record_watermark(k, MarkOrder::Timestamp)
}
_ => Freshness::AlwaysFull {
why: "declares no incremental key, so every run re-reads the whole source"
.into(),
},
}
}
/// **Does this collector re-read its whole source every run?** The one call a pane
/// makes to render "this one re-reads everything" instead of pretending.
pub fn rereads_everything(&self) -> bool {
matches!(self, Freshness::AlwaysFull { .. }) || self.undeclared().is_some()
}
/// **Does a re-run avoid transferring bytes it already has?**
///
/// `false` for a local-only watermark: the body still comes down the wire in full
/// and only the rows are dropped. A pane that showed "incremental" for both would
/// be telling an operator their bandwidth bill is a delta when it is not.
pub fn saves_bytes(&self) -> bool {
// An incomplete declaration saves nothing, whatever kind it names. Found by
// `an_incomplete_declaration_degrades_to_a_full_read_with_its_reason`: a
// `QueryWindow` with a `param` but a BLANK `record_field` answered "yes, this
// saves bytes" while `Plan::of` had already degraded it to a full read — the
// pane would have shown a saving that the run could not make.
if self.undeclared().is_some() {
return false;
}
match self {
Freshness::MutableDocument { validators } => !validators.is_empty(),
Freshness::FileSeries { listing, .. } => !listing.trim().is_empty(),
Freshness::QueryWindow { param, .. } => param.is_some(),
Freshness::AppendOnlyStream { offset_param } => !offset_param.trim().is_empty(),
Freshness::ByteRangeColumnar { object } => !object.trim().is_empty(),
Freshness::AlwaysFull { .. } => false,
}
}
/// **Is the declaration incomplete?** Returns the operator line when a kind was
/// named but the thing it needs to act was left blank — a `FileSeries` with no
/// listing, a `MutableDocument` with no validators. Such a declaration cannot
/// produce a delta, so [`Plan::of`] degrades it to a full read and
/// [`rereads_everything`](Self::rereads_everything) reports it truthfully rather
/// than letting an empty field read as an incremental collector.
pub fn undeclared(&self) -> Option<String> {
match self {
Freshness::MutableDocument { validators } if validators.is_empty() => Some(
"declares a mutable document but names no validator — with neither ETag \
nor Last-Modified there is nothing to send back, so every run refetches \
the whole body"
.into(),
),
Freshness::FileSeries { listing, .. } if listing.trim().is_empty() => Some(
"declares a file series but no listing — with nowhere to enumerate names \
the series cannot be advanced"
.into(),
),
Freshness::QueryWindow { record_field, .. } if record_field.trim().is_empty() => {
Some(
"declares a query window but names no record field — with no per-record \
key there is nothing to take a watermark from"
.into(),
)
}
Freshness::AppendOnlyStream { offset_param } if offset_param.trim().is_empty() => {
Some(
"declares an append-only stream but no offset parameter — the offset \
cannot be sent, so every run reads the log from the start"
.into(),
)
}
Freshness::ByteRangeColumnar { object } if object.trim().is_empty() => Some(
"declares a byte-range columnar object but does not name the object — \
nothing to plan ranges over"
.into(),
),
_ => None,
}
}
/// The one-line label a pane shows in the freshness column.
pub fn label(&self) -> String {
if let Some(u) = self.undeclared() {
return format!("re-reads everything ({u})");
}
match self {
Freshness::MutableDocument { validators } => format!(
"conditional fetch ({})",
validators
.iter()
.map(|v| v.response_header())
.collect::<Vec<_>>()
.join(" + ")
),
Freshness::FileSeries { listing, order, .. } => {
format!("file series from {listing} ({} order)", order.key())
}
Freshness::QueryWindow {
param: Some(p),
granularity,
..
} => format!("query window `{p}=` ({} granularity)", granularity.key()),
Freshness::QueryWindow {
param: None,
record_field,
..
} => format!("record watermark on `{record_field}` (whole body still fetched)"),
Freshness::AppendOnlyStream { offset_param } => {
format!("append-only log, offset `{offset_param}`")
}
Freshness::ByteRangeColumnar { object } => format!("byte ranges of {object}"),
Freshness::AlwaysFull { why } => format!("re-reads everything — {why}"),
}
}
/// The word a pane uses for the mark column's header, so a reader knows what the
/// stored value MEANS: a validator, a filename, a timestamp, an offset, a token.
pub fn mark_noun(&self) -> &'static str {
match self {
Freshness::MutableDocument { .. } => "validator",
Freshness::FileSeries { .. } => "last file",
Freshness::QueryWindow { .. } => "watermark",
Freshness::AppendOnlyStream { .. } => "offset",
Freshness::ByteRangeColumnar { .. } => "version",
Freshness::AlwaysFull { .. } => "—",
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The mark — what one successful run leaves behind
// ─────────────────────────────────────────────────────────────────────────────
/// **What the last successful run of a `(collector, source)` pair read.**
///
/// One record, persisted by the host, holding only facts: what was read, when, how
/// much, and the validators the next conditional request sends back. Everything a pane
/// shows — "unchanged", "3 new rows", "incremental" — is derived from this plus the
/// declaration. There is no stored opinion here.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReadMark {
/// The freshness kind that WROTE this mark ([`Freshness::key`]). Kept so a mark
/// left by a different kind is discarded rather than misread: a filename resumed
/// as a timestamp would silently collect nothing.
pub kind: String,
/// The resume value, in the kind's own terms — the last filename consumed, the
/// high-watermark timestamp, the log offset, the object's version token. Empty
/// for a pure conditional fetch, where the validators ARE the mark.
#[serde(default)]
pub mark: String,
/// The `ETag` the source last served.
#[serde(default)]
pub etag: Option<String>,
/// The `Last-Modified` the source last served. Also the
/// `nornir_parquet_delta::Source::version` token for
/// [`ByteRangeColumnar`](Freshness::ByteRangeColumnar) when no `ETag` exists.
#[serde(default)]
pub last_modified: Option<String>,
/// When this mark was written — RFC-3339 UTC. "Last run" in the pane.
#[serde(default)]
pub at: String,
/// Rows landed by the run that wrote it. "Rows added last run".
#[serde(default)]
pub rows: u64,
/// Bytes fetched by that run. `0` on a `304`, and that zero is the whole point.
#[serde(default)]
pub bytes: u64,
/// `true` when the last run found the source UNCHANGED. The pane says "unchanged"
/// rather than showing a run that looks like it failed.
#[serde(default)]
pub unchanged: bool,
}
impl ReadMark {
/// A mark for a run that landed rows.
pub fn landed(kind: &str, mark: impl Into<String>, rows: u64, bytes: u64) -> Self {
ReadMark {
kind: kind.to_string(),
mark: mark.into(),
at: now_rfc3339(),
rows,
bytes,
unchanged: false,
..Default::default()
}
}
/// Attach the validators a conditional fetch should send back next time.
pub fn with_validators(mut self, etag: Option<String>, last_modified: Option<String>) -> Self {
self.etag = etag;
self.last_modified = last_modified;
self
}
/// **The same mark, re-stamped as "checked, unchanged".** The resume value, the
/// validators and the row count of the last run that actually landed are all
/// KEPT — only `at`, `bytes` and `unchanged` move. An unchanged run must not
/// erase the last real read, or the pane forgets what was ever collected.
pub fn unchanged(&self, bytes: u64) -> Self {
ReadMark {
at: now_rfc3339(),
bytes,
unchanged: true,
..self.clone()
}
}
/// Is this mark usable to resume a collector declaring `f`? A mark written by a
/// different kind is NOT — see [`kind`](ReadMark::kind).
pub fn resumes(&self, f: &Freshness) -> bool {
self.kind == f.key()
}
/// The resume value for `f`, or `None` when the mark is absent, empty or was
/// written by another kind.
pub fn resume_for(this: Option<&ReadMark>, f: &Freshness) -> Option<String> {
this.filter(|m| m.resumes(f))
.map(|m| m.mark.clone())
.filter(|s| !s.is_empty())
}
/// The pane's one-line summary of the last run.
pub fn line(&self, f: &Freshness) -> String {
if self.unchanged {
return format!(
"unchanged at {} ({} {}, {} bytes fetched)",
self.at,
f.mark_noun(),
self.display_mark(),
self.bytes
);
}
format!(
"{} rows at {} ({} {}, {} bytes fetched)",
self.rows,
self.at,
f.mark_noun(),
self.display_mark(),
self.bytes
)
}
/// The mark as a pane shows it — the resume value, or the validator when the
/// validator IS the mark.
pub fn display_mark(&self) -> String {
if !self.mark.is_empty() {
return self.mark.clone();
}
match (&self.etag, &self.last_modified) {
(Some(e), _) => e.clone(),
(None, Some(l)) => l.clone(),
(None, None) => "—".into(),
}
}
}
/// RFC-3339 UTC, seconds resolution, with no chrono dependency: the leaf's allowed
/// dep list has no date crate and one fact does not need one.
fn now_rfc3339() -> String {
let secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0);
let days = secs.div_euclid(86_400);
let tod = secs.rem_euclid(86_400);
// Civil-from-days (Howard Hinnant's algorithm), shifted to a March-based year.
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y + 1 } else { y };
format!(
"{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
tod / 3600,
(tod % 3600) / 60,
tod % 60
)
}
// ─────────────────────────────────────────────────────────────────────────────
// Where marks live
// ─────────────────────────────────────────────────────────────────────────────
/// **The key one mark is stored under**: the collector's name AND the source it read.
///
/// The source is part of the key on purpose. A collector re-pointed at a different
/// endpoint — a different PxWeb table, a different feed URL, a different region cut —
/// has never read the new one, and resuming from the old one's watermark would skip
/// everything before it. Changing the source therefore RESETS the mark by construction,
/// with nothing to remember to clear.
pub fn mark_key(collector: &str, source: &str) -> String {
format!("{collector}\u{1f}{source}")
}
/// **Where a `(collector, source)` mark is kept.** The seam, so the leaf persists
/// nothing itself and the host uses the store it already has.
///
/// [`advance`](MarkStore::advance) is the only writer, and the runner calls it **after**
/// the sink has flushed. See the module docs for why the ordering is the whole safety
/// property.
pub trait MarkStore: Send + Sync {
/// The mark for this pair, or `None` when the pair has never been read.
fn mark(&self, collector: &str, source: &str) -> Option<ReadMark>;
/// Record a new mark. Called ONLY after a successful land.
fn advance(&self, collector: &str, source: &str, mark: &ReadMark) -> anyhow::Result<()>;
/// Forget this pair's mark — the re-cut / reset path, after which the next run is
/// a full read.
fn clear(&self, collector: &str, source: &str) -> anyhow::Result<()>;
/// Every mark held, as `(collector, source, mark)` — what `$KORP_STATE` renders.
fn all(&self) -> Vec<(String, String, ReadMark)>;
}
/// **An in-memory store** — a test's, and the honest answer for a host that has wired
/// no persistence: every run is a first run, which is safe (it re-reads) rather than
/// wrong (it skips).
#[derive(Default)]
pub struct MemMarkStore {
marks: Mutex<BTreeMap<(String, String), ReadMark>>,
}
impl MemMarkStore {
/// An empty store.
pub fn new() -> Self {
Self::default()
}
/// How many marks are held.
pub fn len(&self) -> usize {
self.marks.lock().expect("mark store").len()
}
/// Is it empty?
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl MarkStore for MemMarkStore {
fn mark(&self, collector: &str, source: &str) -> Option<ReadMark> {
self.marks
.lock()
.expect("mark store")
.get(&(collector.to_string(), source.to_string()))
.cloned()
}
fn advance(&self, collector: &str, source: &str, mark: &ReadMark) -> anyhow::Result<()> {
self.marks
.lock()
.expect("mark store")
.insert((collector.to_string(), source.to_string()), mark.clone());
Ok(())
}
fn clear(&self, collector: &str, source: &str) -> anyhow::Result<()> {
self.marks
.lock()
.expect("mark store")
.remove(&(collector.to_string(), source.to_string()));
Ok(())
}
fn all(&self) -> Vec<(String, String, ReadMark)> {
self.marks
.lock()
.expect("mark store")
.iter()
.map(|((c, s), m)| (c.clone(), s.clone(), m.clone()))
.collect()
}
}
/// **The on-disk store** — one JSON file, written whole, under a directory the host
/// owns.
///
/// It is deliberately not a new database. korp already persists its collector settings
/// as JSON under `~/.korp/`, and the daemon already hands every collector a
/// [`disk_dir`](crate::daemon::CollectorCtx::disk_dir) beneath it; this file sits in
/// that same tree, so a `rm -rf ~/.korp/collectors/<name>` — the reset an operator
/// already performs — clears the marks with everything else. `clear` exists for the
/// per-source case.
///
/// Written by rename over a temp file in the same directory, so a crash mid-write
/// leaves the previous marks intact rather than a truncated file. A store whose file is
/// unreadable answers `None` for every pair — the safe direction: re-read, never skip.
pub struct FileMarkStore {
path: std::path::PathBuf,
cache: Mutex<BTreeMap<String, ReadMark>>,
}
impl FileMarkStore {
/// Open (or start) the store at `path`, loading whatever is already there.
pub fn open(path: impl Into<std::path::PathBuf>) -> Self {
let path = path.into();
let cache = std::fs::read_to_string(&path)
.ok()
.and_then(|s| serde_json::from_str::<BTreeMap<String, ReadMark>>(&s).ok())
.unwrap_or_default();
FileMarkStore {
path,
cache: Mutex::new(cache),
}
}
/// The file this store writes.
pub fn path(&self) -> &std::path::Path {
&self.path
}
fn flush(&self, map: &BTreeMap<String, ReadMark>) -> anyhow::Result<()> {
if let Some(dir) = self.path.parent() {
std::fs::create_dir_all(dir)?;
}
let tmp = self.path.with_extension("json.tmp");
std::fs::write(&tmp, serde_json::to_vec_pretty(map)?)?;
std::fs::rename(&tmp, &self.path)?;
Ok(())
}
}
impl MarkStore for FileMarkStore {
fn mark(&self, collector: &str, source: &str) -> Option<ReadMark> {
self.cache
.lock()
.expect("mark store")
.get(&mark_key(collector, source))
.cloned()
}
fn advance(&self, collector: &str, source: &str, mark: &ReadMark) -> anyhow::Result<()> {
let mut map = self.cache.lock().expect("mark store");
map.insert(mark_key(collector, source), mark.clone());
self.flush(&map)
}
fn clear(&self, collector: &str, source: &str) -> anyhow::Result<()> {
let mut map = self.cache.lock().expect("mark store");
map.remove(&mark_key(collector, source));
self.flush(&map)
}
fn all(&self) -> Vec<(String, String, ReadMark)> {
self.cache
.lock()
.expect("mark store")
.iter()
.filter_map(|(k, m)| {
k.split_once('\u{1f}')
.map(|(c, s)| (c.to_string(), s.to_string(), m.clone()))
})
.collect()
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The plan — what THIS run should do
// ─────────────────────────────────────────────────────────────────────────────
/// **What one run should fetch**, derived from the declaration and the stored mark.
///
/// Pure data with no capacity to write anything: a plan can be logged, asserted on and
/// handed to a collector, and it still cannot advance a mark. Only the runner, holding
/// a landed row count, can do that.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "plan", rename_all = "snake_case")]
pub enum Plan {
/// Send these validators; a `304` ends the run as an explained zero.
Conditional {
/// `If-None-Match` value, when an `ETag` is held.
if_none_match: Option<String>,
/// `If-Modified-Since` value, when a `Last-Modified` is held.
if_modified_since: Option<String>,
},
/// Enumerate `listing`, keep the names ordered after `after`, consume those.
Series {
/// The listing to enumerate.
listing: String,
/// The suffix that identifies a member.
name_suffix: String,
/// The last name consumed — `None` on a first run, when every name is new.
after: Option<String>,
/// How the names order.
order: MarkOrder,
},
/// Ask the API for the slice after the watermark, then narrow the records locally.
Window {
/// The query parameter to add, when the server has one.
param: Option<String>,
/// The value to put in it — the mark truncated to the parameter's granularity.
param_value: Option<String>,
/// The watermark itself, as [`Bounds::since`](crate::Bounds::since).
since: Option<String>,
/// The record pointer to filter on.
record_field: String,
/// How the record values order.
order: MarkOrder,
},
/// Read the log from this offset.
Offset {
/// The parameter carrying it.
param: String,
/// The offset — `None` on a first run (read from the start).
offset: Option<String>,
},
/// Plan byte ranges over the object with `nornir-parquet-delta`, holding this
/// version token; a matching token means the local mirror is already current.
ByteRange {
/// The object.
object: String,
/// The version token last seen.
version: Option<String>,
},
/// Read the whole source. The honest plan, with the reason attached.
Full {
/// Why this run reads everything: no declaration, an incomplete one, or a
/// first run with nothing to resume from.
why: String,
},
}
impl Plan {
/// **The plan for this run** — the whole default behaviour of resolving a
/// checkpoint, in one pure function.
///
/// A mark left by a DIFFERENT kind is ignored (see [`ReadMark::kind`]), and an
/// incomplete declaration degrades to [`Plan::Full`] with the declaration's own
/// complaint as the reason. Both are the safe direction: a full read costs
/// bandwidth, a wrong resume costs data.
pub fn of(f: &Freshness, mark: Option<&ReadMark>) -> Plan {
if let Some(why) = f.undeclared() {
return Plan::Full { why };
}
let usable = mark.filter(|m| m.resumes(f));
match f {
Freshness::MutableDocument { validators } => {
let want = |v: Validator| validators.contains(&v);
let etag = usable
.and_then(|m| m.etag.clone())
.filter(|_| want(Validator::ETag));
let lm = usable
.and_then(|m| m.last_modified.clone())
.filter(|_| want(Validator::LastModified));
if etag.is_none() && lm.is_none() {
return Plan::Full {
why: "first run — no validator has been seen yet, so the body must \
be fetched once before a 304 is possible"
.into(),
};
}
Plan::Conditional {
if_none_match: etag,
if_modified_since: lm,
}
}
Freshness::FileSeries {
listing,
name_suffix,
order,
} => Plan::Series {
listing: listing.clone(),
name_suffix: name_suffix.clone(),
after: ReadMark::resume_for(usable, f),
order: *order,
},
Freshness::QueryWindow {
param,
record_field,
order,
granularity,
} => {
let since = ReadMark::resume_for(usable, f);
Plan::Window {
param: param.clone(),
param_value: since
.as_deref()
.filter(|_| param.is_some())
.map(|s| granularity.param_value(s)),
since,
record_field: record_field.clone(),
order: *order,
}
}
Freshness::AppendOnlyStream { offset_param } => Plan::Offset {
param: offset_param.clone(),
offset: ReadMark::resume_for(usable, f),
},
Freshness::ByteRangeColumnar { object } => Plan::ByteRange {
object: object.clone(),
version: usable.and_then(|m| m.etag.clone().or_else(|| m.last_modified.clone())),
},
Freshness::AlwaysFull { why } => Plan::Full { why: why.clone() },
}
}
/// The stable machine key.
pub fn key(&self) -> &'static str {
match self {
Plan::Conditional { .. } => "conditional",
Plan::Series { .. } => "series",
Plan::Window { .. } => "window",
Plan::Offset { .. } => "offset",
Plan::ByteRange { .. } => "byte_range",
Plan::Full { .. } => "full",
}
}
/// **Is this run a delta at all?** `false` for [`Plan::Full`] — the one plan that
/// admits it is re-reading everything.
pub fn is_delta(&self) -> bool {
!matches!(self, Plan::Full { .. })
}
/// **Did this plan actually hold something back?**
///
/// Not the same question as [`is_delta`](Plan::is_delta), and the difference is
/// where an empty run gets laundered. A `QueryWindow` collector on its FIRST run
/// plans `Window { since: None }` — a delta by kind, a full read in fact, because
/// there is no watermark yet to be after. If that run comes back with zero records
/// the source was EMPTY, and "returned no records" describes that rather than
/// explaining it. Only a plan that resumed from a mark or a validator can turn a
/// zero into an honest "nothing new".
///
/// `true` exactly when a mark/validator/offset/version was carried into the
/// request; `false` for [`Plan::Full`] and for every first run of every kind.
pub fn resumed(&self) -> bool {
match self {
Plan::Conditional {
if_none_match,
if_modified_since,
} => if_none_match.is_some() || if_modified_since.is_some(),
Plan::Series { after, .. } => after.is_some(),
Plan::Window { since, .. } => since.is_some(),
Plan::Offset { offset, .. } => offset.is_some(),
Plan::ByteRange { version, .. } => version.is_some(),
Plan::Full { .. } => false,
}
}
/// The [`Bounds`](crate::Bounds) this plan gives the collector — the field that
/// has been `None` on every real run since the seam was declared.
pub fn since(&self) -> Option<String> {
match self {
Plan::Window { since, .. } => since.clone(),
_ => None,
}
}
/// The one-line log the runner writes before it fetches.
pub fn line(&self) -> String {
match self {
Plan::Conditional {
if_none_match,
if_modified_since,
} => format!(
"conditional fetch (if-none-match={}, if-modified-since={})",
if_none_match.as_deref().unwrap_or("—"),
if_modified_since.as_deref().unwrap_or("—")
),
Plan::Series { listing, after, .. } => format!(
"series from {listing}, after {}",
after.as_deref().unwrap_or("(nothing consumed yet)")
),
Plan::Window {
param, param_value, since, ..
} => match (param, param_value) {
(Some(p), Some(v)) => format!("window {p}={v}, records after {}", since.as_deref().unwrap_or("—")),
_ => format!(
"records after {} (whole body fetched — the server takes no lower bound)",
since.as_deref().unwrap_or("(nothing read yet)")
),
},
Plan::Offset { param, offset } => format!(
"log from {param}={}",
offset.as_deref().unwrap_or("(the start)")
),
Plan::ByteRange { object, version } => format!(
"byte ranges of {object} (held version {})",
version.as_deref().unwrap_or("—")
),
Plan::Full { why } => format!("full read — {why}"),
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The outcome
// ─────────────────────────────────────────────────────────────────────────────
/// The sentence an unchanged run puts in
/// [`CollectStats::reason`](crate::CollectStats::reason).
///
/// It must NOT be [`crate::RAGNAROK_UNEXPLAINED`]: an unchanged source is a genuine
/// explanation, so this reads back as
/// [`Ragnarok::ExplainedEmpty`](crate::Ragnarok::ExplainedEmpty) and the run is green.
pub fn unchanged_reason(collector: &str, how: &str) -> String {
format!("{collector}: source unchanged since the last run ({how}) — nothing to land")
}
/// **The prefix every unchanged explanation carries**, so a UI can recognise "unchanged"
/// among the other honest emptinesses (a license gate, a provably empty window) without
/// parsing prose.
pub const UNCHANGED_MARKER: &str = "source unchanged since the last run";
/// Is this a `reason` written by [`unchanged_reason`]?
pub fn is_unchanged_reason(reason: Option<&str>) -> bool {
reason.is_some_and(|r| r.contains(UNCHANGED_MARKER))
}
/// **What one incremental run did** — the runner's verdict, from which the new mark is
/// built.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Delta {
/// The source provably did not change. Nothing was landed and nothing should be.
Unchanged {
/// How it was proven — `304 Not Modified`, `no new file in the listing`, `no
/// record after the watermark`.
how: String,
/// Bytes spent proving it (a conditional request's headers land as 0 body
/// bytes; a local-only watermark still pays for the whole body).
bytes: u64,
},
/// Rows were landed, and here is the new resume value.
Landed {
/// Rows the SINK reported landing.
rows: u64,
/// Bytes fetched.
bytes: u64,
/// The new resume value.
mark: String,
/// The `ETag` seen, if any.
etag: Option<String>,
/// The `Last-Modified` seen, if any.
last_modified: Option<String>,
},
}
impl Delta {
/// **The mark to persist** — `None` when there is nothing to write.
///
/// An [`Unchanged`](Delta::Unchanged) run re-stamps the PREVIOUS mark rather than
/// replacing it, so the pane keeps showing what was last actually read; a run with
/// no previous mark and nothing landed writes nothing at all.
pub fn next_mark(&self, f: &Freshness, previous: Option<&ReadMark>) -> Option<ReadMark> {
match self {
Delta::Unchanged { bytes, .. } => previous
.filter(|m| m.resumes(f))
.map(|m| m.unchanged(*bytes)),
Delta::Landed {
rows,
bytes,
mark,
etag,
last_modified,
} => Some(
ReadMark::landed(f.key(), mark.clone(), *rows, *bytes)
.with_validators(etag.clone(), last_modified.clone()),
),
}
}
/// Rows landed (0 when unchanged).
pub fn rows(&self) -> u64 {
match self {
Delta::Unchanged { .. } => 0,
Delta::Landed { rows, .. } => *rows,
}
}
/// Bytes fetched.
pub fn bytes(&self) -> u64 {
match self {
Delta::Unchanged { bytes, .. } | Delta::Landed { bytes, .. } => *bytes,
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// What a pane shows
// ─────────────────────────────────────────────────────────────────────────────
/// **One collector's freshness row**, as the collector pane renders it and as
/// `$KORP_STATE` carries it. Derived — nothing here is stored.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FreshnessRow {
/// The collector name.
pub collector: String,
/// The source it read.
pub source: String,
/// The declared kind's key ([`Freshness::key`]).
pub kind: String,
/// The human label ([`Freshness::label`]).
pub kind_label: String,
/// What the mark MEANS ([`Freshness::mark_noun`]).
pub mark_noun: String,
/// The last-read mark, or `"—"`.
pub mark: String,
/// When it was last read (RFC-3339), or `""`.
pub last_run: String,
/// Rows added by the last run that landed any.
pub rows_last_run: u64,
/// Bytes fetched by the last run.
pub bytes_last_run: u64,
/// The last run found the source unchanged.
pub unchanged: bool,
/// This collector re-reads its whole source every run
/// ([`Freshness::rereads_everything`]) — the pane says so instead of implying a
/// delta.
pub rereads_everything: bool,
/// A re-run avoids transferring bytes it already has ([`Freshness::saves_bytes`]).
pub saves_bytes: bool,
}
impl FreshnessRow {
/// Build the row from the declaration and whatever mark is stored.
pub fn of(collector: &str, source: &str, f: &Freshness, mark: Option<&ReadMark>) -> Self {
let usable = mark.filter(|m| m.resumes(f));
FreshnessRow {
collector: collector.to_string(),
source: source.to_string(),
kind: f.key().to_string(),
kind_label: f.label(),
mark_noun: f.mark_noun().to_string(),
mark: usable.map(|m| m.display_mark()).unwrap_or_else(|| "—".into()),
last_run: usable.map(|m| m.at.clone()).unwrap_or_default(),
rows_last_run: usable.map(|m| m.rows).unwrap_or(0),
bytes_last_run: usable.map(|m| m.bytes).unwrap_or(0),
unchanged: usable.map(|m| m.unchanged).unwrap_or(false),
rereads_everything: f.rereads_everything(),
saves_bytes: f.saves_bytes(),
}
}
/// The status word the pane puts in the row: `unchanged` is NOT a failure.
pub fn status(&self) -> &'static str {
if self.last_run.is_empty() {
"never read"
} else if self.unchanged {
"unchanged"
} else {
"read"
}
}
/// **One line: how this source changes, and when it was last read.**
///
/// The ONE renderer of this sentence, so the 📡 Collectors row and the 🧵 Pipelines
/// item cannot say different things about the same source (lane EC O7 needed the
/// second, and a second format string would have been the twin).
///
/// It says `re-reads everything` where that is true, because a pane showing
/// "incremental" for a full crawl tells an operator their bandwidth bill is a delta
/// when it is not — and `never read` where there is no mark, which is the fact that
/// makes an empty pipeline explicable.
pub fn line(&self) -> String {
let how = if self.rereads_everything {
"re-reads everything".to_string()
} else {
format!("{} by {}", self.kind_label, self.mark_noun)
};
match self.last_run.is_empty() {
true => format!("{how} · never read"),
false => format!(
"{how} · {} {} · {}{}",
self.status(),
self.last_run,
if self.mark == "—" {
"no mark".to_string()
} else {
format!("{} {}", self.mark_noun, self.mark)
},
if self.rows_last_run > 0 {
format!(" · {} rows", self.rows_last_run)
} else {
String::new()
},
),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
// ── the ordering, red first ──────────────────────────────────────────────
/// **RED:** a lexical order re-consumes `data.2` after `data.10`, forever.
/// **GREEN:** the numeric order does not. Both arms, or the guard proves nothing.
#[test]
fn a_file_series_must_consume_data_2_and_not_re_consume_data_1() {
let names = ["data.1", "data.2", "data.10"];
// RED — the naive answer. `data.10` sorts before `data.2` as text, so after
// consuming `data.10` a lexical run believes `data.2` is still new.
assert!(
MarkOrder::Lexical.is_after("data.2", Some("data.10")),
"the lexical order really does re-offer data.2 — this is the bug"
);
// GREEN — numeric.
assert!(!MarkOrder::Numeric.is_after("data.2", Some("data.10")));
assert!(!MarkOrder::Numeric.is_after("data.1", Some("data.2")));
assert!(MarkOrder::Numeric.is_after("data.10", Some("data.2")));
assert_eq!(
MarkOrder::Numeric.newest(names).as_deref(),
Some("data.10"),
"the newest of the series is data.10, not data.2"
);
// And the plan advances: after data.1, exactly data.2 and data.10 remain.
let after: Vec<&str> = names
.iter()
.copied()
.filter(|n| MarkOrder::Numeric.is_after(n, Some("data.1")))
.collect();
assert_eq!(after, vec!["data.2", "data.10"], "data.1 is not re-consumed");
}
/// **RED:** polisen stamps a single-digit hour, so a lexical watermark drops every
/// event after 09:59. The real string from the live feed, not a made-up one.
#[test]
fn a_ragged_timestamp_must_not_be_compared_lexically() {
let early = "2026-09-03 9:58:00 +02:00";
let late = "2026-09-03 11:23:41 +02:00";
// RED — lexically the 11:23 event is BEFORE the 9:58 one, so a run that parked
// on 9:58 would never see it.
assert!(
!MarkOrder::Lexical.is_after(late, Some(early)),
"lexical really does hide the later event — this is the bug"
);
// GREEN — as a timestamp.
assert!(MarkOrder::Timestamp.is_after(late, Some(early)));
assert!(!MarkOrder::Timestamp.is_after(early, Some(late)));
// …and it is still right for stamps that were already fixed-width.
assert!(MarkOrder::Timestamp.is_after("2026-09-03T11:23:41Z", Some("2026-09-03T09:58:00Z")));
assert!(!MarkOrder::Timestamp.is_after("2026-09-02T23:59:59Z", Some("2026-09-03T00:00:00Z")));
// Nothing beats nothing: a first run takes everything.
assert!(MarkOrder::Timestamp.is_after(early, None));
}
/// The day-granular parameter is asked for the DAY, never the exact mark — a
/// server that takes `DateTime=2026-09-02` returns nothing for a full timestamp.
#[test]
fn a_coarse_window_parameter_is_asked_for_its_own_granularity() {
let mark = "2026-09-02 14:31:07 +02:00";
assert_eq!(WindowGranularity::Exact.param_value(mark), mark);
assert_eq!(WindowGranularity::Day.param_value(mark), "2026-09-02");
assert_eq!(WindowGranularity::Month.param_value(mark), "2026-09");
assert_eq!(WindowGranularity::Year.param_value(mark), "2026");
}
// ── the declaration ──────────────────────────────────────────────────────
/// **An unknown-freshness collector SAYS it re-reads everything.** The (f) case,
/// and the one a UI must never dress up as a delta.
#[test]
fn an_undeclared_collector_says_it_rereads_everything() {
let f = Freshness::derived(None);
assert_eq!(f.key(), "always_full");
assert!(f.rereads_everything(), "it must SAY so");
assert!(!f.saves_bytes());
assert!(f.label().contains("re-reads everything"), "{}", f.label());
assert!(matches!(Plan::of(&f, None), Plan::Full { .. }));
assert!(!Plan::of(&f, None).is_delta());
// A collector WITH a declared record key derives a real (row-level) delta —
// the guard discriminates rather than answering `always_full` to everything.
let g = Freshness::derived(Some("/datetime"));
assert_eq!(g.key(), "query_window");
assert!(!g.rereads_everything());
// …but it is honest that the bytes are NOT saved.
assert!(!g.saves_bytes(), "a local-only watermark still fetches the whole body");
assert!(g.label().contains("whole body still fetched"), "{}", g.label());
// And the default is the honest one, not a claim.
assert_eq!(Freshness::default().key(), "always_full");
assert!(Freshness::default().rereads_everything());
}
/// A kind named but left blank cannot produce a delta, and must not read as one.
#[test]
fn an_incomplete_declaration_degrades_to_a_full_read_with_its_reason() {
let cases = [
Freshness::MutableDocument { validators: vec![] },
Freshness::FileSeries {
listing: " ".into(),
name_suffix: ".parquet".into(),
order: MarkOrder::Numeric,
},
Freshness::QueryWindow {
param: Some("since".into()),
record_field: "".into(),
order: MarkOrder::Lexical,
granularity: WindowGranularity::Exact,
},
Freshness::AppendOnlyStream { offset_param: "".into() },
Freshness::ByteRangeColumnar { object: "".into() },
];
for f in &cases {
assert!(f.undeclared().is_some(), "{} is incomplete", f.key());
assert!(f.rereads_everything(), "{} must admit it re-reads", f.key());
assert!(!f.saves_bytes(), "{} saves nothing", f.key());
match Plan::of(f, None) {
Plan::Full { why } => assert!(!why.is_empty(), "the reason is named"),
other => panic!("{} must degrade to a full read, got {other:?}", f.key()),
}
}
// …and the SAME kinds, completed, do not degrade — the guard is not blind.
assert!(Freshness::conditional().undeclared().is_none());
assert!(Freshness::conditional().saves_bytes());
assert!(
Freshness::FileSeries {
listing: "s3://bucket/prefix/".into(),
name_suffix: ".parquet".into(),
order: MarkOrder::Numeric,
}
.saves_bytes()
);
}
// ── the plan ─────────────────────────────────────────────────────────────
/// A first run has no validator to send, so it fetches once; the SECOND run sends
/// what the first learned. The `304` path cannot exist without this ordering.
#[test]
fn a_conditional_source_fetches_once_then_sends_its_validators() {
let f = Freshness::conditional();
match Plan::of(&f, None) {
Plan::Full { why } => assert!(why.contains("first run"), "{why}"),
other => panic!("a first run has nothing to send: {other:?}"),
}
let mark = ReadMark::landed("mutable_document", "", 500, 190_839).with_validators(
Some("\"abc\"".into()),
Some("Thu, 03 Sep 2026 00:40:00 GMT".into()),
);
match Plan::of(&f, Some(&mark)) {
Plan::Conditional {
if_none_match,
if_modified_since,
} => {
assert_eq!(if_none_match.as_deref(), Some("\"abc\""));
assert_eq!(
if_modified_since.as_deref(),
Some("Thu, 03 Sep 2026 00:40:00 GMT")
);
}
other => panic!("the second run is conditional: {other:?}"),
}
// A source that serves ONLY Last-Modified does not get an If-None-Match
// invented for it.
let lm_only = Freshness::MutableDocument {
validators: vec![Validator::LastModified],
};
match Plan::of(&lm_only, Some(&mark)) {
Plan::Conditional { if_none_match, if_modified_since } => {
assert!(if_none_match.is_none(), "no ETag was declared");
assert!(if_modified_since.is_some());
}
other => panic!("{other:?}"),
}
}
/// **A query window must not re-ask for a slice it has.** The second run carries
/// the watermark, and the coarse parameter carries the watermark's DAY.
#[test]
fn a_query_window_asks_only_for_the_slice_after_its_mark() {
let f = Freshness::query_window(
"DateTime",
"/datetime",
MarkOrder::Timestamp,
WindowGranularity::Day,
);
// First run: nothing to resume from, so the whole current window.
match Plan::of(&f, None) {
Plan::Window { param, param_value, since, .. } => {
assert_eq!(param.as_deref(), Some("DateTime"));
assert!(param_value.is_none(), "nothing to bound by yet");
assert!(since.is_none());
}
other => panic!("{other:?}"),
}
// Second run: the mark bounds it, and `since` is finally NOT None — the field
// that has been None on every real run since the seam was declared.
let mark = ReadMark::landed("query_window", "2026-09-02 14:31:07 +02:00", 500, 190_839);
let p = Plan::of(&f, Some(&mark));
match &p {
Plan::Window { param_value, since, .. } => {
assert_eq!(param_value.as_deref(), Some("2026-09-02"), "the DAY, not the stamp");
assert_eq!(since.as_deref(), Some("2026-09-02 14:31:07 +02:00"));
}
other => panic!("{other:?}"),
}
assert_eq!(p.since().as_deref(), Some("2026-09-02 14:31:07 +02:00"));
assert!(p.is_delta());
}
/// A mark written by one kind must not be resumed by another — a filename read as
/// a timestamp collects nothing, silently.
#[test]
fn a_mark_from_another_kind_is_ignored_not_misread() {
let series_mark = ReadMark::landed("file_series", "data.7", 10, 100);
let window = Freshness::record_watermark("/datetime", MarkOrder::Timestamp);
assert!(!series_mark.resumes(&window));
match Plan::of(&window, Some(&series_mark)) {
Plan::Window { since, .. } => assert!(
since.is_none(),
"a file name is not a watermark — the run must start clean"
),
other => panic!("{other:?}"),
}
// The matching kind DOES resume — the guard is not just refusing everything.
let own = ReadMark::landed("query_window", "2026-09-01T00:00:00Z", 3, 9);
assert!(own.resumes(&window));
assert_eq!(
Plan::of(&window, Some(&own)).since().as_deref(),
Some("2026-09-01T00:00:00Z")
);
}
/// The byte-range kind carries the version token `nornir-parquet-delta` asks for.
#[test]
fn the_byte_range_plan_carries_the_version_token() {
let f = Freshness::ByteRangeColumnar {
object: "s3://nordisk/warehouse/nodes.parquet".into(),
};
assert!(matches!(Plan::of(&f, None), Plan::ByteRange { version: None, .. }));
let mark = ReadMark::landed("byte_range_columnar", "", 0, 0)
.with_validators(Some("gen-42".into()), None);
match Plan::of(&f, Some(&mark)) {
Plan::ByteRange { object, version } => {
assert_eq!(object, "s3://nordisk/warehouse/nodes.parquet");
assert_eq!(version.as_deref(), Some("gen-42"));
}
other => panic!("{other:?}"),
}
}
// ── the mark, and the ordering rule that is the whole safety property ────
/// **A checkpoint must NOT advance when the land fails.** The store is the only
/// writer and the runner calls it only after a flush; this proves the store still
/// holds the OLD mark when nothing was written, so the next run redoes the slice.
#[test]
fn a_failed_land_leaves_the_previous_mark_in_place() {
let store = MemMarkStore::new();
let f = Freshness::record_watermark("/datetime", MarkOrder::Timestamp);
let first = ReadMark::landed(f.key(), "2026-09-01T10:00:00Z", 500, 190_839);
store.advance("polisen-events", "https://polisen.se/api/events", &first).unwrap();
// A run fetches a newer slice and then the LAND fails. Nothing calls advance.
let _would_have_been = ReadMark::landed(f.key(), "2026-09-02T10:00:00Z", 4, 24_715);
// (the runner returns Err before advance — nothing to call here, and that is
// the point: the mark cannot move without a landed row count)
let held = store.mark("polisen-events", "https://polisen.se/api/events").unwrap();
assert_eq!(
held.mark, "2026-09-01T10:00:00Z",
"the mark must NOT have advanced past a slice that never landed"
);
// The next run therefore REDOES that slice rather than skipping it.
assert_eq!(
Plan::of(&f, Some(&held)).since().as_deref(),
Some("2026-09-01T10:00:00Z")
);
// GREEN arm: a successful land DOES move it — the store is not simply frozen.
let second = ReadMark::landed(f.key(), "2026-09-02T10:00:00Z", 4, 24_715);
store.advance("polisen-events", "https://polisen.se/api/events", &second).unwrap();
assert_eq!(
store.mark("polisen-events", "https://polisen.se/api/events").unwrap().mark,
"2026-09-02T10:00:00Z"
);
}
/// An unchanged run keeps what was last actually read — only the check time, the
/// bytes and the flag move.
#[test]
fn an_unchanged_run_restamps_the_mark_without_erasing_it() {
let f = Freshness::conditional();
let landed = ReadMark::landed(f.key(), "", 500, 190_839)
.with_validators(Some("\"v1\"".into()), None);
let next = Delta::Unchanged {
how: "304 Not Modified".into(),
bytes: 0,
}
.next_mark(&f, Some(&landed))
.expect("an unchanged run re-stamps the previous mark");
assert!(next.unchanged);
assert_eq!(next.bytes, 0, "a 304 transfers no body");
assert_eq!(next.rows, 500, "the last REAL read is remembered");
assert_eq!(next.etag.as_deref(), Some("\"v1\""), "the validator survives");
// With no previous mark there is nothing to re-stamp, and nothing is invented.
assert!(
Delta::Unchanged { how: "304".into(), bytes: 0 }.next_mark(&f, None).is_none()
);
}
/// Changing the source resets the mark by construction — the key contains it.
#[test]
fn a_repointed_source_does_not_resume_from_the_old_ones_mark() {
let store = MemMarkStore::new();
let m = ReadMark::landed("query_window", "2026-01-01", 5, 50);
store.advance("pxweb", "https://api.scb.se/OV0104/v1/doris/sv/ssd/BE0101", &m).unwrap();
assert!(store.mark("pxweb", "https://api.scb.se/OV0104/v1/doris/sv/ssd/AM0301").is_none());
assert!(store.mark("pxweb", "https://api.scb.se/OV0104/v1/doris/sv/ssd/BE0101").is_some());
assert_ne!(mark_key("a", "s1"), mark_key("a", "s2"));
assert_ne!(mark_key("a", "s"), mark_key("b", "s"));
// clear() is the explicit per-source reset.
store.clear("pxweb", "https://api.scb.se/OV0104/v1/doris/sv/ssd/BE0101").unwrap();
assert!(store.mark("pxweb", "https://api.scb.se/OV0104/v1/doris/sv/ssd/BE0101").is_none());
}
/// The on-disk store survives a reopen, and clearing it really clears the file.
#[test]
fn the_file_store_round_trips_and_clears() {
let dir = std::env::temp_dir().join(format!("korp-freshness-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let path = dir.join("marks.json");
{
let s = FileMarkStore::open(&path);
assert!(s.mark("c", "u").is_none(), "a fresh store knows nothing");
s.advance("c", "u", &ReadMark::landed("query_window", "m1", 7, 70)).unwrap();
}
{
let s = FileMarkStore::open(&path);
let m = s.mark("c", "u").expect("the mark survived the process");
assert_eq!(m.mark, "m1");
assert_eq!(m.rows, 7);
assert_eq!(s.all().len(), 1);
assert_eq!(s.all()[0].0, "c");
assert_eq!(s.all()[0].1, "u");
s.clear("c", "u").unwrap();
}
assert!(
FileMarkStore::open(&path).mark("c", "u").is_none(),
"clear() is a reset the next run sees"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// An unreadable store answers "never read" rather than pretending to know — the
/// safe direction is a re-read, never a skip.
#[test]
fn a_corrupt_store_file_reads_as_no_marks_at_all() {
let dir = std::env::temp_dir().join(format!("korp-fresh-bad-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("marks.json");
std::fs::write(&path, b"{ not json at all").unwrap();
let s = FileMarkStore::open(&path);
assert!(s.mark("c", "u").is_none());
assert!(s.all().is_empty());
let _ = std::fs::remove_dir_all(&dir);
}
// ── RAGNARÖK ─────────────────────────────────────────────────────────────
/// **An unchanged source is a first-class explained zero, never RED.**
#[test]
fn an_unchanged_source_is_an_explained_zero_and_not_a_failure() {
use crate::{CollectStats, Ragnarok, RAGNAROK_UNEXPLAINED};
let reason = unchanged_reason("usgs-quakes", "304 Not Modified");
assert!(is_unchanged_reason(Some(&reason)));
assert_ne!(reason, RAGNAROK_UNEXPLAINED);
let stats = CollectStats {
rows: 0,
batches: 0,
table: "usgs_quakes".into(),
reason: Some(reason),
};
assert_eq!(stats.ragnarok(), Ragnarok::ExplainedEmpty);
assert!(!stats.is_red(), "an unchanged source must not go red");
assert!(stats.ragnarok_check().is_ok());
// RED arm: the same zero WITHOUT the explanation is still red, and the
// runner's own placeholder does not launder it.
let bare = CollectStats { rows: 0, batches: 0, table: "t".into(), reason: None };
assert!(bare.is_red());
assert!(!is_unchanged_reason(None));
assert!(!is_unchanged_reason(Some(RAGNAROK_UNEXPLAINED)));
}
// ── the pane row ─────────────────────────────────────────────────────────
/// The pane row carries every column the brief asks for, and an unchanged row
/// reads as `unchanged` rather than as a failure.
#[test]
fn the_pane_row_shows_the_mark_the_run_the_rows_and_the_kind() {
let f = Freshness::conditional();
let never = FreshnessRow::of("usgs-quakes", "https://earthquake.usgs.gov/…", &f, None);
assert_eq!(never.status(), "never read");
assert_eq!(never.mark, "—");
assert_eq!(never.rows_last_run, 0);
let landed = ReadMark::landed(f.key(), "", 12, 3_853)
.with_validators(None, Some("Thu, 03 Sep 2026 00:40:00 GMT".into()));
let read = FreshnessRow::of("usgs-quakes", "https://earthquake.usgs.gov/…", &f, Some(&landed));
assert_eq!(read.status(), "read");
assert_eq!(read.rows_last_run, 12);
assert_eq!(read.bytes_last_run, 3_853);
assert_eq!(read.mark, "Thu, 03 Sep 2026 00:40:00 GMT");
assert_eq!(read.mark_noun, "validator");
assert!(read.saves_bytes);
assert!(!read.rereads_everything);
let unchanged = landed.unchanged(0);
let row = FreshnessRow::of("usgs-quakes", "https://earthquake.usgs.gov/…", &f, Some(&unchanged));
assert_eq!(row.status(), "unchanged", "a 304 is not a failure");
assert_eq!(row.bytes_last_run, 0);
assert_eq!(row.rows_last_run, 12, "the last real read is still shown");
// An always-full collector SAYS it re-reads everything.
let full = Freshness::derived(None);
let fr = FreshnessRow::of("kolada-kpi", "https://api.kolada.se/v3/kpi", &full, None);
assert!(fr.rereads_everything);
assert!(!fr.saves_bytes);
assert_eq!(fr.mark_noun, "—");
}
/// Every kind survives JSON, both ways — it travels the wire into `$KORP_STATE`,
/// and a representation that cannot serialize turns the pane into an error.
/// (`requires::Readiness` learned this the hard way with a tagged newtype.)
#[test]
fn every_kind_plan_and_row_round_trips_as_json() {
let kinds = [
Freshness::conditional(),
Freshness::FileSeries {
listing: "https://data.gdeltproject.org/gdeltv2/".into(),
name_suffix: ".export.CSV.zip".into(),
order: MarkOrder::Numeric,
},
Freshness::query_window("DateTime", "/datetime", MarkOrder::Timestamp, WindowGranularity::Day),
Freshness::record_watermark("/time", MarkOrder::Timestamp),
Freshness::AppendOnlyStream { offset_param: "offset".into() },
Freshness::ByteRangeColumnar { object: "s3://b/o.parquet".into() },
Freshness::default(),
];
let mut seen: Vec<&str> = Vec::new();
for f in &kinds {
let js = serde_json::to_string(f).expect("a kind serializes");
let back: Freshness = serde_json::from_str(&js).expect("and comes back");
assert_eq!(&back, f, "lossless: {js}");
assert_eq!(back.label(), f.label(), "the words survive the wire");
seen.push(f.key());
let plan = Plan::of(f, None);
let pjs = serde_json::to_string(&plan).expect("a plan serializes");
assert_eq!(serde_json::from_str::<Plan>(&pjs).unwrap(), plan);
let row = FreshnessRow::of("c", "s", f, None);
let rjs = serde_json::to_string(&row).expect("a row serializes");
assert_eq!(serde_json::from_str::<FreshnessRow>(&rjs).unwrap(), row);
}
// Every declared kind is exercised — a taxonomy with an untested member is a
// taxonomy with an untested member.
for k in Freshness::KEYS {
assert!(seen.contains(k), "kind `{k}` is not covered by this sweep");
}
let m = ReadMark::landed("query_window", "m", 1, 2).with_validators(Some("e".into()), None);
assert_eq!(serde_json::from_str::<ReadMark>(&serde_json::to_string(&m).unwrap()).unwrap(), m);
}
/// The clock helper produces a plausible RFC-3339 stamp — a `at` field that read
/// `1970-01-01` for every row would make "last run" useless.
#[test]
fn the_timestamp_helper_writes_a_real_rfc3339_utc_stamp() {
let s = now_rfc3339();
assert_eq!(s.len(), 20, "YYYY-MM-DDTHH:MM:SSZ — got {s}");
assert!(s.ends_with('Z'));
let year: i32 = s[..4].parse().expect("a year");
assert!((2025..2100).contains(&year), "the clock is live, not epoch: {s}");
assert_eq!(&s[4..5], "-");
assert_eq!(&s[10..11], "T");
let month: u32 = s[5..7].parse().unwrap();
let day: u32 = s[8..10].parse().unwrap();
assert!((1..=12).contains(&month), "{s}");
assert!((1..=31).contains(&day), "{s}");
// A known instant decoded by the same algorithm: 2026-09-03T00:39:50Z is
// 1788482390 seconds after the epoch.
assert!(s.as_str() > "2026-01-01T00:00:00Z", "{s}");
}
}
#[cfg(test)]
mod resumed_tests {
use super::*;
/// **`resumed()` splits the two zeros that look identical.** A `QueryWindow`
/// collector's FIRST run is a delta by kind and a full read in fact — there is no
/// watermark to be after — and calling that plan a delta is precisely how "the
/// source was empty" got laundered into "nothing new since the mark".
///
/// Both halves are asserted for every kind, so a `resumed()` hard-wired to either
/// answer is red.
#[test]
fn resumed_is_true_only_when_the_request_actually_carried_a_mark() {
let window = Freshness::QueryWindow {
param: Some("DateTime".into()),
record_field: "/datetime".into(),
order: MarkOrder::Lexical,
granularity: WindowGranularity::Exact,
};
// First run: no mark held ⇒ a Window with `since: None`, which asks for
// everything and so cannot explain an empty answer.
let first = Plan::of(&window, None);
assert!(matches!(first, Plan::Window { .. }), "kind: {first:?}");
assert!(first.is_delta(), "it IS a delta by kind…");
assert!(!first.resumed(), "…and it resumed from NOTHING: {first:?}");
// Second run: a usable mark ⇒ the same kind, now genuinely bounded.
let mark = ReadMark::landed(window.key(), "2026-09-04T10:00:00Z", 5, 1234);
let second = Plan::of(&window, Some(&mark));
assert!(second.resumed(), "a held watermark is carried: {second:?}");
// AlwaysFull never resumes, in either direction.
let full = Plan::of(
&Freshness::AlwaysFull {
why: "a registry crawl".into(),
},
Some(&mark),
);
assert!(!full.resumed(), "a full read holds nothing back: {full:?}");
assert!(!full.is_delta());
// Every remaining kind, first run vs resumed, on the SAME plan value — so a
// per-variant slip cannot hide behind another variant's correct answer.
assert!(
!Plan::Series {
listing: "l".into(),
name_suffix: ".parquet".into(),
after: None,
order: MarkOrder::Numeric,
}
.resumed()
);
assert!(
Plan::Series {
listing: "l".into(),
name_suffix: ".parquet".into(),
after: Some("data.7".into()),
order: MarkOrder::Numeric,
}
.resumed()
);
assert!(
!Plan::Offset {
param: "from".into(),
offset: None
}
.resumed()
);
assert!(
Plan::Offset {
param: "from".into(),
offset: Some("42".into())
}
.resumed()
);
assert!(
!Plan::ByteRange {
object: "s3://b/k".into(),
version: None
}
.resumed()
);
assert!(
Plan::ByteRange {
object: "s3://b/k".into(),
version: Some("v1".into())
}
.resumed()
);
assert!(
!Plan::Conditional {
if_none_match: None,
if_modified_since: None
}
.resumed()
);
assert!(
Plan::Conditional {
if_none_match: Some("\"e\"".into()),
if_modified_since: None
}
.resumed()
);
assert!(
Plan::Conditional {
if_none_match: None,
if_modified_since: Some("Thu, 04 Sep 2026 10:00:00 GMT".into())
}
.resumed()
);
}
}
#[cfg(test)]
mod row_line_tests {
use super::*;
/// **`line()` must SAY the three things a pane cannot infer**, and it is asserted on
/// rows where a wrong implementation gives a different answer for each.
#[test]
fn the_row_line_says_how_when_and_what_was_read() {
let window = Freshness::QueryWindow {
param: Some("DateTime".into()),
record_field: "/datetime".into(),
order: MarkOrder::Timestamp,
granularity: WindowGranularity::Day,
};
// Never read: the fact that makes an empty pipeline explicable rather than
// suspicious. No timestamp is invented.
let never = FreshnessRow::of("polisen-events", "https://polisen.se/api/events", &window, None);
let l = never.line();
assert!(l.ends_with("never read"), "{l}");
assert!(!l.contains("20"), "no invented date: {l}");
// Read, with a mark: the cadence, the status, the instant and the watermark.
let mut mark = ReadMark::landed(window.key(), "2026-09-04T08:00:00Z", 500, 4096);
mark.at = "2026-09-04T08:00:01Z".into();
let read = FreshnessRow::of("polisen-events", "https://polisen.se/api/events", &window, Some(&mark));
let l = read.line();
assert!(l.contains("2026-09-04T08:00:01Z"), "WHEN it last ran: {l}");
assert!(l.contains("2026-09-04T08:00:00Z"), "the watermark itself: {l}");
assert!(l.contains("read"), "{l}");
assert!(l.contains("500 rows"), "{l}");
assert!(!l.contains("re-reads everything"), "a window is a delta: {l}");
// Unchanged is NOT a failure and must read as its own status.
let unchanged = mark.unchanged(0);
let u = FreshnessRow::of("polisen-events", "https://polisen.se/api/events", &window, Some(&unchanged));
assert!(u.line().contains("unchanged"), "{}", u.line());
// A full crawl SAYS so — a pane showing "incremental" for it tells an operator
// their bandwidth bill is a delta when it is not.
let full = Freshness::AlwaysFull { why: "a national catalogue".into() };
let fr = FreshnessRow::of("dcat-ap-se", "https://dataportal.se/", &full, None);
assert!(fr.line().starts_with("re-reads everything"), "{}", fr.line());
assert_ne!(fr.line(), read.line(), "two different sources cannot render alike");
}
}