faucet_cli/backfill/
plan.rs

1//! Pure window/unit planning for `faucet backfill` — boundary parsing, range
2//! chunking (timezone/DST-correct), `${backfill.*}` token substitution, and
3//! the stable range hash the progress marker is keyed by. No I/O.
4
5use crate::error::{CliError, CliResult};
6use chrono::{DateTime, Duration, FixedOffset, TimeZone, Utc};
7use serde_json::Value;
8
9/// Hard ceiling on planned units — a tiny `--window` over a huge range is a
10/// config error, not a workload.
11pub const MAX_UNITS: usize = 10_000;
12/// Above this many units a loud warning is emitted (but planning proceeds).
13pub const WARN_UNITS: usize = 1_000;
14
15/// One independent, resumable slice of the backfill range.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct BackfillUnit {
18    /// Stable unit id — the UTC start instant, compact (`20260601T000000Z`).
19    /// Doubles as the state-key suffix (`{name}::backfill::{id}`).
20    pub id: String,
21    /// Half-open window start (inclusive). Carried in the range's timezone
22    /// offset so `${now.*}` renders local dates.
23    pub start: DateTime<FixedOffset>,
24    /// Half-open window end (exclusive).
25    pub end: DateTime<FixedOffset>,
26}
27
28/// Parse a `--window` duration: `45s`, `30m`, `6h`, `1d`, `1w` (or a bare
29/// integer = seconds). Must be positive.
30pub fn parse_window(s: &str) -> CliResult<Duration> {
31    let s = s.trim();
32    let err = || {
33        CliError::Config(format!(
34            "'{s}' is not a valid window — use e.g. 45s, 30m, 6h, 1d, 1w"
35        ))
36    };
37    let (num, unit) = match s.chars().last() {
38        Some(c) if c.is_ascii_digit() => (s, "s"),
39        Some(c) => (&s[..s.len() - c.len_utf8()], &s[s.len() - c.len_utf8()..]),
40        None => return Err(err()),
41    };
42    let n: i64 = num.parse().map_err(|_| err())?;
43    if n <= 0 {
44        return Err(CliError::Config(format!(
45            "window '{s}' must be a positive duration"
46        )));
47    }
48    let dur = match unit {
49        "s" => Duration::seconds(n),
50        "m" => Duration::minutes(n),
51        "h" => Duration::hours(n),
52        "d" => Duration::days(n),
53        "w" => Duration::weeks(n),
54        _ => return Err(err()),
55    };
56    Ok(dur)
57}
58
59/// Parse a `--from` / `--to` boundary: RFC3339 (`2026-06-01T00:00:00Z`) or a
60/// bare date (`2026-06-01`, interpreted as midnight in `tz`). A date that
61/// falls in a DST gap resolves to the earliest valid instant.
62pub fn parse_boundary(s: &str, tz: chrono_tz::Tz) -> CliResult<DateTime<FixedOffset>> {
63    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
64        return Ok(dt.with_timezone(&tz).fixed_offset());
65    }
66    if let Ok(date) = s.parse::<chrono::NaiveDate>() {
67        let midnight = date
68            .and_hms_opt(0, 0, 0)
69            .ok_or_else(|| CliError::Config(format!("'{s}' has no valid midnight in {tz}")))?;
70        let local = tz
71            .from_local_datetime(&midnight)
72            .earliest()
73            .ok_or_else(|| {
74                CliError::Config(format!("'{s}' midnight does not exist in {tz} (DST gap)"))
75            })?;
76        return Ok(local.fixed_offset());
77    }
78    Err(CliError::Config(format!(
79        "'{s}' is not RFC3339 (2026-06-01T00:00:00Z) or a date (2026-06-01)"
80    )))
81}
82
83/// Chunk `[from, to)` into contiguous half-open windows of `window` (the last
84/// window truncated at `to`). `window: None` = the whole range as one unit.
85/// Window arithmetic is absolute (instants), so units never gap or overlap —
86/// including across DST transitions; boundaries are re-rendered in `tz` so
87/// `${now.*}` tokens see local wall-clock time.
88pub fn plan_windows(
89    from: DateTime<FixedOffset>,
90    to: DateTime<FixedOffset>,
91    window: Option<Duration>,
92    tz: chrono_tz::Tz,
93) -> CliResult<Vec<BackfillUnit>> {
94    if from >= to {
95        return Err(CliError::Config(format!(
96            "--from ({from}) must be before --to ({to})"
97        )));
98    }
99    let mut units = Vec::new();
100    let mut cursor = from.with_timezone(&Utc);
101    let end = to.with_timezone(&Utc);
102    let step = window.unwrap_or_else(|| end - cursor);
103    while cursor < end {
104        if units.len() >= MAX_UNITS {
105            return Err(CliError::Config(format!(
106                "the range would produce more than {MAX_UNITS} units with this --window — \
107                 use a larger window"
108            )));
109        }
110        let unit_end = (cursor + step).min(end);
111        units.push(BackfillUnit {
112            id: cursor.format("%Y%m%dT%H%M%SZ").to_string(),
113            start: cursor.with_timezone(&tz).fixed_offset(),
114            end: unit_end.with_timezone(&tz).fixed_offset(),
115        });
116        cursor = unit_end;
117    }
118    Ok(units)
119}
120
121/// Substitute `${backfill.*}` tokens in every string leaf of `value`:
122/// `start` / `end` (RFC3339), `start_date` / `end_date` (`YYYY-MM-DD`, local),
123/// `start_unix` / `end_unix` (epoch seconds), `unit` (the unit id). An
124/// unrecognized `${backfill.*}` token is a typo — typed error.
125pub fn substitute_unit_tokens(value: &mut Value, unit: &BackfillUnit) -> CliResult<()> {
126    match value {
127        Value::String(s) => {
128            *s = substitute_in_str(s, unit)?;
129            Ok(())
130        }
131        Value::Array(a) => a
132            .iter_mut()
133            .try_for_each(|v| substitute_unit_tokens(v, unit)),
134        Value::Object(m) => m
135            .values_mut()
136            .try_for_each(|v| substitute_unit_tokens(v, unit)),
137        _ => Ok(()),
138    }
139}
140
141fn substitute_in_str(input: &str, unit: &BackfillUnit) -> CliResult<String> {
142    const PREFIX: &str = "${backfill.";
143    let mut out = String::with_capacity(input.len());
144    let mut rest = input;
145    while let Some(pos) = rest.find(PREFIX) {
146        out.push_str(&rest[..pos]);
147        let after = &rest[pos + PREFIX.len()..];
148        let close = after.find('}').ok_or_else(|| {
149            CliError::Config(format!("unterminated ${{backfill.…}} token in '{input}'"))
150        })?;
151        let token = &after[..close];
152        let rendered = match token {
153            "start" => unit.start.to_rfc3339(),
154            "end" => unit.end.to_rfc3339(),
155            "start_date" => unit.start.format("%Y-%m-%d").to_string(),
156            "end_date" => unit.end.format("%Y-%m-%d").to_string(),
157            "start_unix" => unit.start.timestamp().to_string(),
158            "end_unix" => unit.end.timestamp().to_string(),
159            "unit" => unit.id.clone(),
160            other => {
161                return Err(CliError::Config(format!(
162                    "unknown token ${{backfill.{other}}} — supported: start, end, start_date, \
163                     end_date, start_unix, end_unix, unit"
164                )));
165            }
166        };
167        out.push_str(&rendered);
168        rest = &after[close + 1..];
169    }
170    out.push_str(rest);
171    Ok(out)
172}
173
174/// Deterministic 64-bit FNV-1a hash of the range descriptor, hex-encoded.
175/// Keys the progress marker so `--resume` finds the same backfill across
176/// process restarts (std's `DefaultHasher` is randomly seeded — unusable).
177pub fn range_hash(descriptor: &str) -> String {
178    const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
179    const PRIME: u64 = 0x0000_0100_0000_01b3;
180    let mut hash = OFFSET;
181    for b in descriptor.as_bytes() {
182        hash ^= u64::from(*b);
183        hash = hash.wrapping_mul(PRIME);
184    }
185    format!("{hash:016x}")
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191    use serde_json::json;
192
193    fn tz(name: &str) -> chrono_tz::Tz {
194        name.parse().unwrap()
195    }
196
197    #[test]
198    fn window_durations_parse() {
199        assert_eq!(parse_window("45s").unwrap(), Duration::seconds(45));
200        assert_eq!(parse_window("30m").unwrap(), Duration::minutes(30));
201        assert_eq!(parse_window("6h").unwrap(), Duration::hours(6));
202        assert_eq!(parse_window("1d").unwrap(), Duration::days(1));
203        assert_eq!(parse_window("2w").unwrap(), Duration::weeks(2));
204        assert_eq!(parse_window("3600").unwrap(), Duration::seconds(3600));
205        assert!(parse_window("0d").is_err());
206        assert!(parse_window("-1h").is_err());
207        assert!(parse_window("soon").is_err());
208        assert!(parse_window("1y").is_err());
209    }
210
211    #[test]
212    fn boundaries_parse_rfc3339_and_dates() {
213        let utc = tz("UTC");
214        let dt = parse_boundary("2026-06-01T12:30:00Z", utc).unwrap();
215        assert_eq!(dt.to_rfc3339(), "2026-06-01T12:30:00+00:00");
216        // A bare date is midnight in the given timezone.
217        let ny = tz("America/New_York");
218        let dt = parse_boundary("2026-06-01", ny).unwrap();
219        assert_eq!(dt.to_rfc3339(), "2026-06-01T00:00:00-04:00");
220        assert!(parse_boundary("yesterday", utc).is_err());
221    }
222
223    #[test]
224    fn thirty_one_days_one_day_window_is_31_units() {
225        // The acceptance-criteria example: a 31-day June-July range with a 1d
226        // window plans exactly 31 units.
227        let utc = tz("UTC");
228        let from = parse_boundary("2026-06-01", utc).unwrap();
229        let to = parse_boundary("2026-07-02", utc).unwrap();
230        let units = plan_windows(from, to, Some(Duration::days(1)), utc).unwrap();
231        assert_eq!(units.len(), 31);
232        assert_eq!(units[0].id, "20260601T000000Z");
233        assert_eq!(units[0].start.to_rfc3339(), "2026-06-01T00:00:00+00:00");
234        assert_eq!(units[0].end.to_rfc3339(), "2026-06-02T00:00:00+00:00");
235        // Contiguous half-open windows: each start equals the previous end.
236        for w in units.windows(2) {
237            assert_eq!(w[0].end, w[1].start);
238        }
239        assert_eq!(units[30].end.to_rfc3339(), "2026-07-02T00:00:00+00:00");
240    }
241
242    #[test]
243    fn last_window_truncates_at_to() {
244        let utc = tz("UTC");
245        let from = parse_boundary("2026-06-01T00:00:00Z", utc).unwrap();
246        let to = parse_boundary("2026-06-01T05:30:00Z", utc).unwrap();
247        let units = plan_windows(from, to, Some(Duration::hours(2)), utc).unwrap();
248        assert_eq!(units.len(), 3);
249        assert_eq!(units[2].start.to_rfc3339(), "2026-06-01T04:00:00+00:00");
250        assert_eq!(units[2].end.to_rfc3339(), "2026-06-01T05:30:00+00:00");
251    }
252
253    #[test]
254    fn no_window_is_a_single_unit() {
255        let utc = tz("UTC");
256        let from = parse_boundary("2026-06-01", utc).unwrap();
257        let to = parse_boundary("2026-07-01", utc).unwrap();
258        let units = plan_windows(from, to, None, utc).unwrap();
259        assert_eq!(units.len(), 1);
260        assert_eq!(units[0].start, from);
261        assert_eq!(units[0].end, to);
262    }
263
264    #[test]
265    fn dst_transition_produces_no_gap_or_overlap() {
266        // US spring-forward 2026: March 8, 02:00 EST → 03:00 EDT. Absolute
267        // 1-day windows stay contiguous; local render shows the offset flip.
268        let ny = tz("America/New_York");
269        let from = parse_boundary("2026-03-07", ny).unwrap();
270        let to = parse_boundary("2026-03-10T00:00:00-04:00", ny).unwrap();
271        let units = plan_windows(from, to, Some(Duration::days(1)), ny).unwrap();
272        for w in units.windows(2) {
273            assert_eq!(w[0].end, w[1].start, "no gap/overlap across DST");
274        }
275        // First window starts EST (-05:00); a later one renders EDT (-04:00).
276        assert!(units[0].start.to_rfc3339().ends_with("-05:00"));
277        assert!(units.last().unwrap().end.to_rfc3339().ends_with("-04:00"));
278    }
279
280    #[test]
281    fn rejects_inverted_range_and_unit_explosion() {
282        let utc = tz("UTC");
283        let from = parse_boundary("2026-06-02", utc).unwrap();
284        let to = parse_boundary("2026-06-01", utc).unwrap();
285        assert!(plan_windows(from, to, None, utc).is_err());
286
287        let from = parse_boundary("2020-01-01", utc).unwrap();
288        let to = parse_boundary("2026-01-01", utc).unwrap();
289        let err = plan_windows(from, to, Some(Duration::minutes(1)), utc).unwrap_err();
290        assert!(err.to_string().contains("larger window"), "{err}");
291    }
292
293    #[test]
294    fn tokens_substitute_in_nested_config() {
295        let utc = tz("UTC");
296        let unit = BackfillUnit {
297            id: "20260601T000000Z".into(),
298            start: parse_boundary("2026-06-01T00:00:00Z", utc).unwrap(),
299            end: parse_boundary("2026-06-02T00:00:00Z", utc).unwrap(),
300        };
301        let mut cfg = json!({
302            "query": "SELECT * FROM t WHERE ts >= '${backfill.start}' AND ts < '${backfill.end}'",
303            "nested": { "path": "dt=${backfill.start_date}/part-${backfill.unit}.jsonl" },
304            "unix": ["${backfill.start_unix}", "${backfill.end_unix}"],
305            "count": 3,
306        });
307        substitute_unit_tokens(&mut cfg, &unit).unwrap();
308        assert_eq!(
309            cfg["query"],
310            "SELECT * FROM t WHERE ts >= '2026-06-01T00:00:00+00:00' AND ts < '2026-06-02T00:00:00+00:00'"
311        );
312        assert_eq!(
313            cfg["nested"]["path"],
314            "dt=2026-06-01/part-20260601T000000Z.jsonl"
315        );
316        assert_eq!(cfg["unix"][0], "1780272000");
317        assert_eq!(cfg["count"], 3);
318    }
319
320    #[test]
321    fn unknown_or_unterminated_token_is_a_typed_error() {
322        let utc = tz("UTC");
323        let unit = BackfillUnit {
324            id: "u".into(),
325            start: parse_boundary("2026-06-01", utc).unwrap(),
326            end: parse_boundary("2026-06-02", utc).unwrap(),
327        };
328        let mut bad = json!({"q": "${backfill.begin}"});
329        let err = substitute_unit_tokens(&mut bad, &unit).unwrap_err();
330        assert!(err.to_string().contains("backfill.begin"), "{err}");
331        let mut unterminated = json!({"q": "${backfill.start"});
332        assert!(substitute_unit_tokens(&mut unterminated, &unit).is_err());
333    }
334
335    #[test]
336    fn range_hash_is_stable_and_distinct() {
337        let a = range_hash("2026-06-01|2026-07-01|1d");
338        assert_eq!(a, range_hash("2026-06-01|2026-07-01|1d"), "deterministic");
339        assert_ne!(a, range_hash("2026-06-01|2026-07-01|6h"));
340        assert_eq!(a.len(), 16);
341    }
342}
faucet_cli/backfill/plan.rs

faucet_cli/backfill/
plan.rs
