faucet_cli/backfill/
plan.rs

1//! Backfill-specific planning: `${backfill.*}` token substitution and the stable
2//! range hash the progress marker is keyed by.
3//!
4//! The window planning itself lives in [`crate::chunking`], shared with the
5//! `partition:` block (#479) so both have one implementation and one set of
6//! boundary tests. It is re-exported here unchanged, so every existing call site
7//! and test keeps working against the moved code — which is what demonstrates
8//! the move did not alter backfill's behaviour.
9
10use crate::error::{CliError, CliResult};
11use serde_json::Value;
12
13pub use crate::chunking::{
14    MAX_UNITS, TimeChunk as BackfillUnit, WARN_UNITS, WindowStep, parse_boundary, parse_window,
15    plan_windows,
16};
17
18/// Substitute `${backfill.*}` tokens in every string leaf of `value`:
19/// `start` / `end` (RFC3339), `start_date` / `end_date` (`YYYY-MM-DD`, local),
20/// `start_unix` / `end_unix` (epoch seconds), `unit` (the unit id). An
21/// unrecognized `${backfill.*}` token is a typo — typed error.
22pub fn substitute_unit_tokens(value: &mut Value, unit: &BackfillUnit) -> CliResult<()> {
23    match value {
24        Value::String(s) => {
25            *s = substitute_in_str(s, unit)?;
26            Ok(())
27        }
28        Value::Array(a) => a
29            .iter_mut()
30            .try_for_each(|v| substitute_unit_tokens(v, unit)),
31        Value::Object(m) => m
32            .values_mut()
33            .try_for_each(|v| substitute_unit_tokens(v, unit)),
34        _ => Ok(()),
35    }
36}
37
38fn substitute_in_str(input: &str, unit: &BackfillUnit) -> CliResult<String> {
39    const PREFIX: &str = "${backfill.";
40    let mut out = String::with_capacity(input.len());
41    let mut rest = input;
42    while let Some(pos) = rest.find(PREFIX) {
43        out.push_str(&rest[..pos]);
44        let after = &rest[pos + PREFIX.len()..];
45        let close = after.find('}').ok_or_else(|| {
46            CliError::Config(format!("unterminated ${{backfill.…}} token in '{input}'"))
47        })?;
48        let token = &after[..close];
49        let rendered = match token {
50            "start" => unit.start.to_rfc3339(),
51            "end" => unit.end.to_rfc3339(),
52            "start_date" => unit.start.format("%Y-%m-%d").to_string(),
53            "end_date" => unit.end.format("%Y-%m-%d").to_string(),
54            "start_unix" => unit.start.timestamp().to_string(),
55            "end_unix" => unit.end.timestamp().to_string(),
56            "unit" => unit.id.clone(),
57            other => {
58                return Err(CliError::Config(format!(
59                    "unknown token ${{backfill.{other}}} — supported: start, end, start_date, \
60                     end_date, start_unix, end_unix, unit"
61                )));
62            }
63        };
64        out.push_str(&rendered);
65        rest = &after[close + 1..];
66    }
67    out.push_str(rest);
68    Ok(out)
69}
70
71/// Deterministic 64-bit FNV-1a hash of the range descriptor, hex-encoded.
72/// Keys the progress marker so `--resume` finds the same backfill across
73/// process restarts (std's `DefaultHasher` is randomly seeded — unusable).
74pub fn range_hash(descriptor: &str) -> String {
75    const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
76    const PRIME: u64 = 0x0000_0100_0000_01b3;
77    let mut hash = OFFSET;
78    for b in descriptor.as_bytes() {
79        hash ^= u64::from(*b);
80        hash = hash.wrapping_mul(PRIME);
81    }
82    format!("{hash:016x}")
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88    use serde_json::json;
89
90    fn tz(name: &str) -> chrono_tz::Tz {
91        name.parse().unwrap()
92    }
93
94    #[test]
95    fn tokens_substitute_in_nested_config() {
96        let utc = tz("UTC");
97        let unit = BackfillUnit {
98            id: "20260601T000000Z".into(),
99            start: parse_boundary("2026-06-01T00:00:00Z", utc).unwrap(),
100            end: parse_boundary("2026-06-02T00:00:00Z", utc).unwrap(),
101        };
102        let mut cfg = json!({
103            "query": "SELECT * FROM t WHERE ts >= '${backfill.start}' AND ts < '${backfill.end}'",
104            "nested": { "path": "dt=${backfill.start_date}/part-${backfill.unit}.jsonl" },
105            "unix": ["${backfill.start_unix}", "${backfill.end_unix}"],
106            "count": 3,
107        });
108        substitute_unit_tokens(&mut cfg, &unit).unwrap();
109        assert_eq!(
110            cfg["query"],
111            "SELECT * FROM t WHERE ts >= '2026-06-01T00:00:00+00:00' AND ts < '2026-06-02T00:00:00+00:00'"
112        );
113        assert_eq!(
114            cfg["nested"]["path"],
115            "dt=2026-06-01/part-20260601T000000Z.jsonl"
116        );
117        assert_eq!(cfg["unix"][0], "1780272000");
118        assert_eq!(cfg["count"], 3);
119    }
120
121    #[test]
122    fn unknown_or_unterminated_token_is_a_typed_error() {
123        let utc = tz("UTC");
124        let unit = BackfillUnit {
125            id: "u".into(),
126            start: parse_boundary("2026-06-01", utc).unwrap(),
127            end: parse_boundary("2026-06-02", utc).unwrap(),
128        };
129        let mut bad = json!({"q": "${backfill.begin}"});
130        let err = substitute_unit_tokens(&mut bad, &unit).unwrap_err();
131        assert!(err.to_string().contains("backfill.begin"), "{err}");
132        let mut unterminated = json!({"q": "${backfill.start"});
133        assert!(substitute_unit_tokens(&mut unterminated, &unit).is_err());
134    }
135
136    #[test]
137    fn range_hash_is_stable_and_distinct() {
138        let a = range_hash("2026-06-01|2026-07-01|1d");
139        assert_eq!(a, range_hash("2026-06-01|2026-07-01|1d"), "deterministic");
140        assert_ne!(a, range_hash("2026-06-01|2026-07-01|6h"));
141        assert_eq!(a.len(), 16);
142    }
143}
faucet_cli/backfill/plan.rs

faucet_cli/backfill/
plan.rs
