faucet_cli/backfill/
spec.rs

1//! Serde config types for the optional top-level `backfill:` block.
2//!
3//! The block holds **defaults** for `faucet backfill` (window size,
4//! concurrency, timezone); the actual range always comes from the command
5//! line (`--from/--to` or `--from-bookmark/--to-bookmark`). Ignored by
6//! `faucet run` (like `schedule:` / `replication:`).
7
8use crate::error::{CliError, CliResult};
9use schemars::JsonSchema;
10use serde::{Deserialize, Serialize};
11
12/// Top-level `backfill:` defaults block.
13#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema, PartialEq)]
14#[serde(deny_unknown_fields)]
15pub struct BackfillSpec {
16    /// Default window chunk when `--window` is not passed — a duration like
17    /// `1d`, `6h`, `30m`, `45s`, or `1w`. Omitted = the whole range runs as a
18    /// single unit.
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub window: Option<String>,
21    /// Default max concurrently-running window units when `--concurrency` is
22    /// not passed. Defaults to 1 (sequential).
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub concurrency: Option<usize>,
25    /// IANA timezone (e.g. `America/New_York`) in which date boundaries like
26    /// `--from 2026-06-01` are interpreted and `${now.*}` tokens render.
27    /// Defaults to UTC.
28    #[serde(default, skip_serializing_if = "Option::is_none")]
29    pub timezone: Option<String>,
30}
31
32impl BackfillSpec {
33    /// Fail-fast validation of the defaults block, run by `faucet validate`
34    /// and again by `faucet backfill` before any execution. `source_configs`
35    /// are the serialized source configs of the pipeline's root rows — a
36    /// `backfill:` block on a pipeline whose source references no
37    /// `${backfill.*}` / `${now.*}` scoping token would replay identical data
38    /// into every window, so it is rejected here (bookmark-mode backfills
39    /// don't need the block at all).
40    pub fn validate(&self, source_configs: &[String]) -> CliResult<()> {
41        if let Some(w) = &self.window {
42            crate::backfill::plan::parse_window(w)?;
43        }
44        if self.concurrency == Some(0) {
45            return Err(CliError::Config(
46                "backfill.concurrency must be at least 1".into(),
47            ));
48        }
49        if let Some(tz) = &self.timezone {
50            parse_timezone(tz)?;
51        }
52        if !source_configs.is_empty() && !source_configs.iter().any(|c| has_scoping_tokens(c)) {
53            return Err(CliError::Config(
54                "the config has a `backfill:` block but no source config references a \
55                 `${backfill.start}` / `${backfill.end}` / `${now.*}` token — every window \
56                 would replay identical data. Scope the source to the window (e.g. \
57                 `query: SELECT * FROM t WHERE updated_at >= '${backfill.start}' AND \
58                 updated_at < '${backfill.end}'`), or drop the block and use \
59                 `faucet backfill --from-bookmark` instead"
60                    .into(),
61            ));
62        }
63        Ok(())
64    }
65}
66
67/// Whether a serialized source config references a window-scoping token.
68pub fn has_scoping_tokens(serialized_config: &str) -> bool {
69    serialized_config.contains("${backfill.") || serialized_config.contains("${now.")
70}
71
72/// Parse an IANA timezone name.
73pub fn parse_timezone(name: &str) -> CliResult<chrono_tz::Tz> {
74    name.parse::<chrono_tz::Tz>().map_err(|_| {
75        CliError::Config(format!(
76            "'{name}' is not a valid IANA timezone (e.g. UTC, America/New_York)"
77        ))
78    })
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84
85    #[test]
86    fn parses_full_block() {
87        let yaml = "window: 1d\nconcurrency: 4\ntimezone: America/New_York\n";
88        let spec: BackfillSpec = serde_yaml::from_str(yaml).unwrap();
89        assert_eq!(spec.window.as_deref(), Some("1d"));
90        assert_eq!(spec.concurrency, Some(4));
91        spec.validate(&["${backfill.start}".into()]).unwrap();
92    }
93
94    #[test]
95    fn rejects_unknown_field() {
96        assert!(serde_yaml::from_str::<BackfillSpec>("bogus: 1\n").is_err());
97    }
98
99    #[test]
100    fn rejects_bad_window_concurrency_timezone() {
101        let spec = BackfillSpec {
102            window: Some("soon".into()),
103            ..Default::default()
104        };
105        assert!(spec.validate(&[]).is_err());
106        let spec = BackfillSpec {
107            concurrency: Some(0),
108            ..Default::default()
109        };
110        assert!(spec.validate(&[]).is_err());
111        let spec = BackfillSpec {
112            timezone: Some("Mars/Olympus".into()),
113            ..Default::default()
114        };
115        assert!(spec.validate(&[]).is_err());
116    }
117
118    #[test]
119    fn rejects_unscoped_source_with_block() {
120        let spec = BackfillSpec::default();
121        let err = spec
122            .validate(&[r#"{"query":"SELECT * FROM t"}"#.into()])
123            .unwrap_err();
124        assert!(err.to_string().contains("${backfill.start}"), "{err}");
125        // A `${now.*}`-scoped source passes.
126        spec.validate(&[r#"{"prefix":"dt=${now.date}/"}"#.into()])
127            .unwrap();
128        // No sources to check (empty) passes.
129        spec.validate(&[]).unwrap();
130    }
131
132    #[test]
133    fn timezone_parses() {
134        parse_timezone("UTC").unwrap();
135        parse_timezone("America/New_York").unwrap();
136        assert!(parse_timezone("Nowhere").is_err());
137    }
138}
faucet_cli/backfill/spec.rs

faucet_cli/backfill/
spec.rs
