faucet_core/
window.rs

1//! In-run datetime window slicing for forward incremental (#527).
2//!
3//! [`ReplicationBind`](crate::ReplicationBind) (#513) binds only a single *lower*
4//! bound (`?since=<bookmark>`); [`faucet backfill`](https://…) (#282) windows only
5//! a *bounded historical* range. Neither bounds each request of the ordinary
6//! forward-incremental run.
7//!
8//! Many APIs — analytics / ads / reporting feeds especially — require **both** a
9//! lower and an upper bound and **cap the span** (e.g. reject a range over 30 or
10//! 90 days). Against those, an unbounded `?since=<bookmark>` either errors or,
11//! worse, silently truncates. Window slicing bounds each request to a rolling
12//! `[start, end)` interval between the persisted bookmark and `now`, iterating the
13//! windows within a single run and persisting the window boundary as the bookmark
14//! at each step so the run is resumable mid-sweep. This is parity with Airbyte's
15//! `DatetimeBasedCursor` (`start_datetime` / `end_datetime` / `step` /
16//! `cursor_granularity` / `lookback_window`).
17//!
18//! The enumeration is a pure function ([`enumerate_windows`]); a source injects
19//! the rendered boundaries into its requests via the [`WindowBind`]s (the window
20//! analogue of [`ReplicationBind`](crate::ReplicationBind)).
21
22use crate::FaucetError;
23use crate::replication::{BindFormat, BindTarget, format_instant};
24use chrono::{DateTime, Duration, Utc};
25use schemars::JsonSchema;
26use serde::{Deserialize, Serialize};
27
28/// The placeholder replaced by the formatted window boundary inside a
29/// [`WindowBind::template`]. The `lower` bind renders the window **start**, the
30/// `upper` bind renders the window **end**.
31pub const WINDOW_PLACEHOLDER: &str = "${window}";
32
33fn default_window_template() -> String {
34    WINDOW_PLACEHOLDER.to_owned()
35}
36
37/// Default [`WindowSpec::max_windows`]: a runaway backstop, far above any real
38/// sweep. On a first run against years of history at a small `step`, the sweep is
39/// truncated here and the next run resumes from the last window.
40pub const DEFAULT_MAX_WINDOWS: usize = 10_000;
41
42fn default_max_windows() -> usize {
43    DEFAULT_MAX_WINDOWS
44}
45
46/// One half-open `[start, end)` slice of the replication timeline.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub struct Window {
49    /// Inclusive lower bound.
50    pub start: DateTime<Utc>,
51    /// Exclusive upper bound.
52    pub end: DateTime<Utc>,
53}
54
55/// Injects a rendered window boundary into the outgoing request — the window
56/// analogue of [`ReplicationBind`](crate::ReplicationBind). Reuses the same
57/// [`BindTarget`] placement and [`BindFormat`] formatting.
58#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
59#[serde(deny_unknown_fields)]
60pub struct WindowBind {
61    /// Where to place the rendered boundary (query param / header / body field /
62    /// path placeholder).
63    #[serde(default)]
64    pub into: BindTarget,
65    /// The parameter / header / body-field / path-placeholder name.
66    pub name: String,
67    /// Template rendered with [`WINDOW_PLACEHOLDER`] (`${window}`) replaced by the
68    /// formatted boundary. Defaults to the bare `${window}`; set e.g.
69    /// `"gte|${window}"` or `"[${window} TO *]"`.
70    #[serde(default = "default_window_template")]
71    pub template: String,
72    /// How to format the boundary before substitution.
73    #[serde(default)]
74    pub format: BindFormat,
75}
76
77impl WindowBind {
78    /// Validate the bind at config-load time. `side` names the field for errors
79    /// (`"lower"` / `"upper"`).
80    pub fn validate(&self, side: &str) -> Result<(), FaucetError> {
81        if self.name.trim().is_empty() {
82            return Err(FaucetError::Config(format!(
83                "window slicing: `{side}.name` must not be empty"
84            )));
85        }
86        if !self.template.contains(WINDOW_PLACEHOLDER) {
87            return Err(FaucetError::Config(format!(
88                "window slicing: `{side}.template` must contain the `{WINDOW_PLACEHOLDER}` placeholder"
89            )));
90        }
91        Ok(())
92    }
93
94    /// Render the bind for a concrete boundary instant.
95    pub fn render(&self, boundary: DateTime<Utc>) -> String {
96        let formatted = format_instant(boundary, self.format);
97        self.template.replace(WINDOW_PLACEHOLDER, &formatted)
98    }
99}
100
101/// Declarative in-run datetime window slicing (#527).
102#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
103#[serde(deny_unknown_fields)]
104pub struct WindowSpec {
105    /// Window size — `45s` / `30m` / `6h` / `30d`, or a bare integer (= seconds).
106    /// Absolute UTC durations (`d` = 24h); calendar/DST-correct windows are a
107    /// [`faucet backfill`] concern, not the incremental cursor.
108    pub step: String,
109    /// Lower-bound bind, rendered with the window **start**.
110    pub lower: WindowBind,
111    /// Upper-bound bind, rendered with the window **end**.
112    pub upper: WindowBind,
113    /// Subtract this from each window's *rendered* upper bound so `[start, end]`
114    /// is non-overlapping for inclusive-inclusive APIs (Airbyte
115    /// `cursor_granularity`). Same grammar as `step`. The **persisted bookmark is
116    /// always the true half-open boundary**, so resume never gaps or overlaps —
117    /// only the value sent to the server is adjusted.
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub granularity: Option<String>,
120    /// Re-scan this much *before* the bookmark on the first window, to catch
121    /// late-arriving updates without a full replay. Same grammar as `step`.
122    #[serde(default, skip_serializing_if = "Option::is_none")]
123    pub lookback: Option<String>,
124    /// Safety cap on the number of windows enumerated in one run. On overflow the
125    /// sweep is truncated (logged, never silently), and the next run resumes from
126    /// the last window's end.
127    #[serde(default = "default_max_windows")]
128    pub max_windows: usize,
129}
130
131/// Parse a [`WindowSpec`] duration string (`step` / `granularity` / `lookback`)
132/// into an absolute [`chrono::Duration`]: `45s` / `30m` / `6h` / `30d` (`d` =
133/// 24h), or a bare integer (= seconds). Must be positive.
134pub fn parse_step(s: &str) -> Result<Duration, FaucetError> {
135    let s = s.trim();
136    let err = || {
137        FaucetError::Config(format!(
138            "window slicing: '{s}' is not a valid duration — use e.g. 45s, 30m, 6h, 30d"
139        ))
140    };
141    let (num, unit) = match s.chars().last() {
142        Some(c) if c.is_ascii_digit() => (s, "s"),
143        Some(c) => (&s[..s.len() - c.len_utf8()], &s[s.len() - c.len_utf8()..]),
144        None => return Err(err()),
145    };
146    let n: i64 = num.parse().map_err(|_| err())?;
147    if n <= 0 {
148        return Err(FaucetError::Config(format!(
149            "window slicing: duration '{s}' must be positive"
150        )));
151    }
152    Ok(match unit {
153        "s" => Duration::seconds(n),
154        "m" => Duration::minutes(n),
155        "h" => Duration::hours(n),
156        "d" => Duration::days(n),
157        _ => return Err(err()),
158    })
159}
160
161impl WindowSpec {
162    /// Validate the whole spec at config-load time.
163    pub fn validate(&self) -> Result<(), FaucetError> {
164        parse_step(&self.step)?;
165        if let Some(g) = &self.granularity {
166            parse_step(g)?;
167        }
168        if let Some(l) = &self.lookback {
169            parse_step(l)?;
170        }
171        self.lower.validate("lower")?;
172        self.upper.validate("upper")?;
173        if self.max_windows == 0 {
174            return Err(FaucetError::Config(
175                "window slicing: `max_windows` must be greater than zero".to_owned(),
176            ));
177        }
178        Ok(())
179    }
180
181    /// The parsed `step` duration.
182    pub fn step_duration(&self) -> Result<Duration, FaucetError> {
183        parse_step(&self.step)
184    }
185
186    /// The parsed `granularity` duration, if any.
187    pub fn granularity_duration(&self) -> Result<Option<Duration>, FaucetError> {
188        self.granularity.as_deref().map(parse_step).transpose()
189    }
190
191    /// The parsed `lookback` duration, if any.
192    pub fn lookback_duration(&self) -> Result<Option<Duration>, FaucetError> {
193        self.lookback.as_deref().map(parse_step).transpose()
194    }
195
196    /// The rendered lower-bound value for a window (the window **start**).
197    pub fn render_lower(&self, w: &Window) -> String {
198        self.lower.render(w.start)
199    }
200
201    /// The rendered upper-bound value for a window, applying `granularity` (the
202    /// window **end**, minus `granularity` if set, for inclusive-inclusive APIs).
203    pub fn render_upper(&self, w: &Window) -> Result<String, FaucetError> {
204        let end = match self.granularity_duration()? {
205            Some(g) => w.end - g,
206            None => w.end,
207        };
208        Ok(self.upper.render(end))
209    }
210}
211
212/// Enumerate contiguous half-open `[start, end)` windows from `start` (minus
213/// `lookback`) up to `now`, each `step` wide (the last clamped to `now`).
214///
215/// Returns `(windows, truncated)`: an empty vec when `start >= now` (a no-op
216/// run); `truncated = true` when the sweep hit `max_windows` before reaching
217/// `now` (the caller logs it — the next run resumes from the last window's end,
218/// which is the persisted bookmark).
219pub fn enumerate_windows(
220    start: DateTime<Utc>,
221    now: DateTime<Utc>,
222    step: Duration,
223    lookback: Option<Duration>,
224    max_windows: usize,
225) -> (Vec<Window>, bool) {
226    let mut cur = match lookback {
227        Some(lb) => start - lb,
228        None => start,
229    };
230    let mut out = Vec::new();
231    let mut truncated = false;
232    while cur < now {
233        if out.len() >= max_windows {
234            truncated = true;
235            break;
236        }
237        let end = std::cmp::min(cur + step, now);
238        // `parse_step` guarantees a positive step, so `cur + step > cur`; the
239        // clamp to `now` also keeps `end > cur` because the loop guard is
240        // `cur < now`. This guard is belt-and-braces against a degenerate clock.
241        if end <= cur {
242            break;
243        }
244        out.push(Window { start: cur, end });
245        cur = end;
246    }
247    (out, truncated)
248}
249
250#[cfg(test)]
251mod tests {
252    use super::*;
253    use chrono::TimeZone;
254    use serde_json::json;
255
256    fn dt(s: &str) -> DateTime<Utc> {
257        DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)
258    }
259
260    #[test]
261    fn parse_step_units() {
262        assert_eq!(parse_step("45s").unwrap(), Duration::seconds(45));
263        assert_eq!(parse_step("30m").unwrap(), Duration::minutes(30));
264        assert_eq!(parse_step("6h").unwrap(), Duration::hours(6));
265        assert_eq!(parse_step("30d").unwrap(), Duration::days(30));
266        assert_eq!(parse_step("3600").unwrap(), Duration::seconds(3600));
267    }
268
269    #[test]
270    fn parse_step_rejects_bad() {
271        assert!(parse_step("0d").is_err());
272        assert!(parse_step("-1h").is_err());
273        assert!(parse_step("").is_err());
274        assert!(parse_step("10y").is_err());
275        assert!(parse_step("abc").is_err());
276    }
277
278    #[test]
279    fn enumerate_contiguous_half_open() {
280        let (ws, trunc) = enumerate_windows(
281            dt("2024-01-01T00:00:00Z"),
282            dt("2024-01-04T00:00:00Z"),
283            Duration::days(1),
284            None,
285            100,
286        );
287        assert!(!trunc);
288        assert_eq!(ws.len(), 3);
289        assert_eq!(ws[0].start, dt("2024-01-01T00:00:00Z"));
290        assert_eq!(ws[0].end, dt("2024-01-02T00:00:00Z"));
291        // Half-open: window N's end equals window N+1's start (no gap, no overlap).
292        assert_eq!(ws[0].end, ws[1].start);
293        assert_eq!(ws[2].end, dt("2024-01-04T00:00:00Z"));
294    }
295
296    #[test]
297    fn last_window_clamps_to_now() {
298        let (ws, _) = enumerate_windows(
299            dt("2024-01-01T00:00:00Z"),
300            dt("2024-01-02T06:00:00Z"),
301            Duration::days(1),
302            None,
303            100,
304        );
305        assert_eq!(ws.len(), 2);
306        assert_eq!(ws[1].start, dt("2024-01-02T00:00:00Z"));
307        assert_eq!(ws[1].end, dt("2024-01-02T06:00:00Z")); // clamped, not +1 day
308    }
309
310    #[test]
311    fn empty_when_start_at_or_after_now() {
312        let (ws, trunc) = enumerate_windows(
313            dt("2024-06-01T00:00:00Z"),
314            dt("2024-06-01T00:00:00Z"),
315            Duration::days(1),
316            None,
317            100,
318        );
319        assert!(ws.is_empty());
320        assert!(!trunc);
321    }
322
323    #[test]
324    fn lookback_extends_the_first_window_backwards() {
325        let (ws, _) = enumerate_windows(
326            dt("2024-01-02T00:00:00Z"),
327            dt("2024-01-03T00:00:00Z"),
328            Duration::days(1),
329            Some(Duration::hours(6)),
330            100,
331        );
332        // First window now starts 6h before the bookmark.
333        assert_eq!(ws[0].start, dt("2024-01-01T18:00:00Z"));
334    }
335
336    #[test]
337    fn max_windows_truncates_and_flags() {
338        let (ws, trunc) = enumerate_windows(
339            dt("2024-01-01T00:00:00Z"),
340            dt("2024-12-31T00:00:00Z"),
341            Duration::days(1),
342            None,
343            5,
344        );
345        assert_eq!(ws.len(), 5);
346        assert!(trunc);
347        // The next run resumes from the last window's end.
348        assert_eq!(ws[4].end, dt("2024-01-06T00:00:00Z"));
349    }
350
351    #[test]
352    fn render_lower_and_upper_with_granularity() {
353        let spec = WindowSpec {
354            step: "1d".into(),
355            lower: WindowBind {
356                into: BindTarget::Query,
357                name: "start".into(),
358                template: "${window}".into(),
359                format: BindFormat::Date,
360            },
361            upper: WindowBind {
362                into: BindTarget::Query,
363                name: "end".into(),
364                template: "${window}".into(),
365                format: BindFormat::Date,
366            },
367            granularity: Some("1d".into()),
368            lookback: None,
369            max_windows: DEFAULT_MAX_WINDOWS,
370        };
371        let w = Window {
372            start: dt("2024-01-01T00:00:00Z"),
373            end: dt("2024-01-02T00:00:00Z"),
374        };
375        assert_eq!(spec.render_lower(&w), "2024-01-01");
376        // Upper is end - granularity (inclusive-inclusive): 2024-01-01, not -02.
377        assert_eq!(spec.render_upper(&w).unwrap(), "2024-01-01");
378    }
379
380    #[test]
381    fn render_template_and_epoch_format() {
382        let bind = WindowBind {
383            into: BindTarget::Query,
384            name: "since".into(),
385            template: "gte|${window}".into(),
386            format: BindFormat::EpochS,
387        };
388        let ts = Utc.timestamp_opt(1_700_000_000, 0).unwrap();
389        assert_eq!(bind.render(ts), "gte|1700000000");
390    }
391
392    #[test]
393    fn validate_catches_misconfig() {
394        let ok = WindowSpec {
395            step: "1d".into(),
396            lower: WindowBind {
397                into: BindTarget::Query,
398                name: "start".into(),
399                template: "${window}".into(),
400                format: BindFormat::Iso8601,
401            },
402            upper: WindowBind {
403                into: BindTarget::Query,
404                name: "end".into(),
405                template: "${window}".into(),
406                format: BindFormat::Iso8601,
407            },
408            granularity: None,
409            lookback: None,
410            max_windows: DEFAULT_MAX_WINDOWS,
411        };
412        ok.validate().unwrap();
413
414        let mut bad_step = ok.clone();
415        bad_step.step = "0d".into();
416        assert!(bad_step.validate().is_err());
417
418        let mut empty_name = ok.clone();
419        empty_name.lower.name = "  ".into();
420        assert!(empty_name.validate().is_err());
421
422        let mut no_placeholder = ok.clone();
423        no_placeholder.upper.template = "fixed".into();
424        assert!(no_placeholder.validate().is_err());
425
426        let mut zero_windows = ok.clone();
427        zero_windows.max_windows = 0;
428        assert!(zero_windows.validate().is_err());
429    }
430
431    #[test]
432    fn spec_deserializes_from_yaml_shape() {
433        let v = json!({
434            "step": "30d",
435            "lower": {"into": "query", "name": "start_date", "format": "date"},
436            "upper": {"into": "query", "name": "end_date", "format": "date"},
437            "lookback": "1d"
438        });
439        let spec: WindowSpec = serde_json::from_value(v).unwrap();
440        assert_eq!(spec.step, "30d");
441        assert_eq!(spec.lower.template, WINDOW_PLACEHOLDER); // defaulted
442        assert_eq!(spec.max_windows, DEFAULT_MAX_WINDOWS); // defaulted
443        spec.validate().unwrap();
444    }
445}
faucet_core/window.rs

faucet_core/
window.rs
