faucet_cli/backfill/
state.rs

1//! Durable progress marker for `faucet backfill` — one JSON document per
2//! backfill range at `{name}::__backfill__::{range_hash}` in the pipeline's
3//! state store, recording each unit's terminal outcome so `--resume` re-runs
4//! only failed/pending units. Kept separate from every live bookmark key
5//! (`{name}::{row}`) and every unit's scoped key (`{name}::backfill::{unit}`).
6
7use crate::backfill::plan::BackfillUnit;
8use crate::error::{CliError, CliResult};
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11use std::collections::BTreeMap;
12
13/// Terminal outcome of one unit.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case", tag = "status")]
16pub enum UnitOutcome {
17    Done,
18    Failed {
19        #[serde(default)]
20        error: String,
21    },
22}
23
24/// The durable backfill marker.
25#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
26pub struct BackfillState {
27    /// Human-readable range descriptor (also the hash input) — lets an
28    /// operator identify the backfill when inspecting the state store.
29    pub descriptor: String,
30    /// Unit id → terminal outcome. Pending units are absent.
31    #[serde(default)]
32    pub units: BTreeMap<String, UnitOutcome>,
33}
34
35impl BackfillState {
36    pub fn new(descriptor: impl Into<String>) -> Self {
37        Self {
38            descriptor: descriptor.into(),
39            units: BTreeMap::new(),
40        }
41    }
42
43    pub fn to_value(&self) -> CliResult<Value> {
44        serde_json::to_value(self)
45            .map_err(|e| CliError::Internal(format!("backfill state serialize: {e}")))
46    }
47
48    pub fn from_value(v: Value) -> CliResult<Self> {
49        serde_json::from_value(v)
50            .map_err(|e| CliError::Config(format!("backfill state parse: {e}")))
51    }
52
53    pub fn mark_done(&mut self, unit: &str) {
54        self.units.insert(unit.to_string(), UnitOutcome::Done);
55    }
56
57    pub fn mark_failed(&mut self, unit: &str, error: impl Into<String>) {
58        self.units.insert(
59            unit.to_string(),
60            UnitOutcome::Failed {
61                error: error.into(),
62            },
63        );
64    }
65
66    pub fn is_done(&self, unit: &str) -> bool {
67        matches!(self.units.get(unit), Some(UnitOutcome::Done))
68    }
69
70    pub fn done_count(&self) -> usize {
71        self.units
72            .values()
73            .filter(|o| matches!(o, UnitOutcome::Done))
74            .count()
75    }
76
77    pub fn failed_count(&self) -> usize {
78        self.units.len() - self.done_count()
79    }
80}
81
82/// State key holding a range's progress marker.
83pub fn marker_key(pipeline_name: &str, range_hash: &str) -> String {
84    format!("{pipeline_name}::__backfill__::{range_hash}")
85}
86
87/// State key a unit's pipeline invocation reads/advances — the executor's key
88/// for a root node whose id is `backfill::{unit}` (namespaced away from the
89/// live `{name}::{row}` key, so the forward sync's bookmark is never touched).
90pub fn unit_state_key(pipeline_name: &str, unit_id: &str) -> String {
91    crate::executor::build_state_key(pipeline_name, &unit_row_id(unit_id), None)
92}
93
94/// The synthesized row id for a unit's node.
95pub fn unit_row_id(unit_id: &str) -> String {
96    format!("backfill::{unit_id}")
97}
98
99/// Split the plan into (to-run, already-done) against a loaded marker: done
100/// units are skipped, failed and pending units run. A fresh marker runs
101/// everything.
102pub fn split_remaining(
103    plan: Vec<BackfillUnit>,
104    state: &BackfillState,
105) -> (Vec<BackfillUnit>, usize) {
106    let (done, todo): (Vec<_>, Vec<_>) = plan.into_iter().partition(|u| state.is_done(&u.id));
107    (todo, done.len())
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113    use crate::backfill::plan::{parse_boundary, plan_windows};
114
115    fn units(n: usize) -> Vec<BackfillUnit> {
116        let utc: chrono_tz::Tz = "UTC".parse().unwrap();
117        let from = parse_boundary("2026-06-01", utc).unwrap();
118        let to = parse_boundary(&format!("2026-06-{:02}", n + 1), utc).unwrap();
119        plan_windows(
120            from,
121            to,
122            Some(crate::backfill::plan::WindowStep::Days(1)),
123            utc,
124        )
125        .unwrap()
126    }
127
128    #[test]
129    fn marker_round_trips() {
130        let mut s = BackfillState::new("2026-06-01|2026-07-01|1d");
131        s.mark_done("20260601T000000Z");
132        s.mark_failed("20260602T000000Z", "connection refused");
133        let back = BackfillState::from_value(s.to_value().unwrap()).unwrap();
134        assert_eq!(back, s);
135        assert_eq!(back.done_count(), 1);
136        assert_eq!(back.failed_count(), 1);
137    }
138
139    #[test]
140    fn keys_are_valid_and_namespaced() {
141        let marker = marker_key("orders", "0123456789abcdef");
142        assert_eq!(marker, "orders::__backfill__::0123456789abcdef");
143        faucet_core::state::validate_state_key(&marker).unwrap();
144
145        let unit = unit_state_key("orders", "20260601T000000Z");
146        assert_eq!(unit, "orders::backfill::20260601T000000Z");
147        faucet_core::state::validate_state_key(&unit).unwrap();
148
149        // The invariant that protects the live bookmark: no unit key ever
150        // equals the forward-sync key for any plausible row id.
151        assert_ne!(
152            unit,
153            crate::executor::build_state_key("orders", "default", None)
154        );
155    }
156
157    #[test]
158    fn split_remaining_skips_done_retries_failed() {
159        let plan = units(3);
160        let mut state = BackfillState::new("d");
161        state.mark_done(&plan[0].id);
162        state.mark_failed(&plan[1].id, "boom");
163        let (todo, skipped) = split_remaining(plan.clone(), &state);
164        assert_eq!(skipped, 1);
165        let ids: Vec<&str> = todo.iter().map(|u| u.id.as_str()).collect();
166        assert_eq!(ids, vec![plan[1].id.as_str(), plan[2].id.as_str()]);
167
168        // Failed→done transition removes it from the failed set.
169        state.mark_done(&plan[1].id);
170        assert_eq!(state.failed_count(), 0);
171    }
172
173    #[test]
174    fn fresh_marker_runs_everything() {
175        let plan = units(2);
176        let (todo, skipped) = split_remaining(plan.clone(), &BackfillState::new("d"));
177        assert_eq!(todo, plan);
178        assert_eq!(skipped, 0);
179    }
180}
faucet_cli/backfill/state.rs

faucet_cli/backfill/
state.rs
