faucet_cli/serve/handlers/
backfill.rs

1//! `POST /v1/backfill` — submit a windowed backfill (#282) over the control
2//! plane. The range is planned server-side (same pure planner as `faucet
3//! backfill`) and **one tracked run is submitted per window unit** through the
4//! standard runner path, so every unit gets the full run lifecycle: history
5//! record, SSE logs, cancel, `timeout_secs`, cluster pull-balancing, and —
6//! when the config carries `shard: { count }` — Mode-B sharding tracked via
7//! `shard_progress` (a single wide window becomes one sharded run).
8//!
9//! Per unit, the submitted document is rewritten: `${backfill.*}` tokens are
10//! substituted, the pipeline `name` is suffixed (`{name}-backfill-{unit}`) so
11//! unit state keys never touch the forward-sync bookmark, and `delivery` is
12//! forced to `at_least_once` (pair with `write_mode: upsert` for idempotent
13//! replays). Deterministic idempotency keys (`backfill:{hash}:{unit}`) make
14//! re-POSTing the same backfill replay-safe: already-submitted units are
15//! replayed, unsubmitted ones proceed — the API-level resume.
16//!
17//! Bookmark-range backfills (`--from-bookmark`) are CLI-only: they seed
18//! scoped state and wrap the source in-process, which a fire-and-forget
19//! submission cannot do.
20
21use crate::backfill::plan::{parse_boundary, parse_window, range_hash, substitute_unit_tokens};
22use crate::backfill::spec::{has_scoping_tokens, parse_timezone};
23use crate::serve::error::ServeError;
24use crate::serve::load::load_submission;
25use crate::serve::rbac::AuthContext;
26use crate::serve::runner::{self, ConfigFormatWire, SubmitRequest};
27use crate::serve::state::ServerState;
28use axum::Json;
29use axum::extract::{Extension, State};
30use axum::http::StatusCode;
31use serde::{Deserialize, Serialize};
32use serde_json::Value;
33use std::collections::BTreeMap;
34
35/// `POST /v1/backfill` request body.
36#[derive(Debug, Deserialize)]
37pub struct BackfillSubmitRequest {
38    /// The pipeline config document (same body shape as `POST /v1/runs`).
39    pub config: String,
40    #[serde(default)]
41    pub config_format: ConfigFormatWire,
42    /// Window start (inclusive): RFC3339 or a date (midnight in `timezone`).
43    pub from: String,
44    /// Window end (exclusive): RFC3339 or a date.
45    pub to: String,
46    /// Chunk duration (`45s`, `30m`, `6h`, `1d`, `1w`). Defaults to the
47    /// config's `backfill.window`; omitted = one unit for the whole range.
48    #[serde(default)]
49    pub window: Option<String>,
50    /// IANA timezone for date boundaries / `${now.*}` rendering. Defaults to
51    /// the config's `backfill.timezone`, else UTC.
52    #[serde(default)]
53    pub timezone: Option<String>,
54    /// Base run name; unit runs are named `{name}-backfill-{unit}`. Defaults
55    /// to the config's `name`.
56    #[serde(default)]
57    pub name: Option<String>,
58    /// Labels merged onto every unit run (plus the generated
59    /// `backfill` / `backfill_unit` labels).
60    #[serde(default)]
61    pub labels: BTreeMap<String, String>,
62    /// Per-unit run timeout.
63    #[serde(default)]
64    pub timeout_secs: Option<u64>,
65}
66
67/// One planned unit's submission outcome.
68#[derive(Debug, Serialize)]
69pub struct BackfillUnitRun {
70    pub unit: String,
71    pub start: String,
72    pub end: String,
73    /// `submitted` | `not_submitted` (queue full — re-POST to continue).
74    pub status: String,
75    #[serde(skip_serializing_if = "Option::is_none")]
76    pub run_id: Option<String>,
77    #[serde(skip_serializing_if = "Option::is_none")]
78    pub error: Option<String>,
79}
80
81/// `POST /v1/backfill` success body (202).
82#[derive(Debug, Serialize)]
83pub struct BackfillSubmitResponse {
84    /// Stable range hash — the `backfill` label on every unit run.
85    pub backfill: String,
86    pub descriptor: String,
87    pub planned: usize,
88    pub submitted: usize,
89    pub units: Vec<BackfillUnitRun>,
90}
91
92/// `POST /v1/backfill` → 202 with one tracked run per window unit.
93pub async fn submit_backfill(
94    State(state): State<ServerState>,
95    Extension(actor): Extension<AuthContext>,
96    Json(req): Json<BackfillSubmitRequest>,
97) -> Result<(StatusCode, Json<BackfillSubmitResponse>), ServeError> {
98    // Validate the config loads/expands and gate the window scoping exactly
99    // like the CLI: every root's source must reference a `${backfill.*}` /
100    // `${now.*}` token or each unit would replay identical data.
101    let loaded = load_submission(
102        &req.config,
103        req.config_format.into(),
104        state.default_base().as_ref(),
105    )
106    .await?;
107    let unscoped: Vec<&str> = loaded
108        .nodes
109        .iter()
110        .filter(|n| matches!(n.role, crate::expand::NodeRole::Root))
111        .filter(|n| !has_scoping_tokens(&n.source.config.to_string()))
112        .map(|n| n.id.as_str())
113        .collect();
114    if !unscoped.is_empty() {
115        return Err(ServeError::BadConfig(format!(
116            "root row(s) {} are not scoped to the backfill window — their source configs \
117             reference no `${{backfill.start}}` / `${{backfill.end}}` / `${{now.*}}` token, \
118             so every window would replay identical data (bookmark-positioned backfills \
119             are CLI-only: `faucet backfill --from-bookmark`)",
120            unscoped.join(", ")
121        )));
122    }
123
124    let spec = loaded.cfg.backfill.clone().unwrap_or_default();
125    let tz = match req.timezone.as_deref().or(spec.timezone.as_deref()) {
126        Some(name) => parse_timezone(name).map_err(|e| ServeError::BadConfig(e.to_string()))?,
127        None => chrono_tz::Tz::UTC,
128    };
129    let window = match req.window.as_deref().or(spec.window.as_deref()) {
130        Some(w) => Some(parse_window(w).map_err(|e| ServeError::BadConfig(e.to_string()))?),
131        None => None,
132    };
133    let from = parse_boundary(&req.from, tz).map_err(|e| ServeError::BadConfig(e.to_string()))?;
134    let to = parse_boundary(&req.to, tz).map_err(|e| ServeError::BadConfig(e.to_string()))?;
135    let units = crate::backfill::plan::plan_windows(from, to, window, tz)
136        .map_err(|e| ServeError::BadConfig(e.to_string()))?;
137
138    let base_name = req
139        .name
140        .clone()
141        .or_else(|| loaded.cfg.name.clone())
142        .unwrap_or_else(|| "pipeline".to_string());
143    let descriptor = format!(
144        "time|{}|{}|{}|{base_name}",
145        from.to_rfc3339(),
146        to.to_rfc3339(),
147        window
148            .map(|w| w.to_string())
149            .unwrap_or_else(|| "whole".into()),
150    );
151    let hash = range_hash(&descriptor);
152
153    // Parse the RAW submitted document once; each unit rewrites a copy. The
154    // raw body (not the default-merged config) is submitted so the runner's
155    // own merge/validate path applies per unit.
156    let doc: Value = serde_yaml::from_str(&req.config)
157        .map_err(|e| ServeError::BadConfig(format!("config is not valid YAML/JSON: {e}")))?;
158
159    crate::serve::audit::write(
160        &state,
161        &actor,
162        "backfill.submit",
163        None,
164        Some(hash.clone()),
165        "ok",
166    )
167    .await;
168
169    let planned = units.len();
170    let mut reports = Vec::with_capacity(planned);
171    let mut submitted = 0usize;
172    let mut queue_full = false;
173    for unit in units {
174        if queue_full {
175            reports.push(BackfillUnitRun {
176                unit: unit.id.clone(),
177                start: unit.start.to_rfc3339(),
178                end: unit.end.to_rfc3339(),
179                status: "not_submitted".into(),
180                run_id: None,
181                error: Some("run queue full — re-POST the same request to continue".into()),
182            });
183            continue;
184        }
185        let unit_name = format!("{base_name}-backfill-{}", unit.id);
186        let unit_doc = rewrite_unit_doc(&doc, &unit, &unit_name)
187            .map_err(|e| ServeError::BadConfig(e.to_string()))?;
188        let mut labels = req.labels.clone();
189        labels.insert("backfill".into(), hash.clone());
190        labels.insert("backfill_unit".into(), unit.id.clone());
191        let submit = SubmitRequest {
192            config: unit_doc,
193            config_format: ConfigFormatWire::Yaml,
194            name: Some(unit_name),
195            labels,
196            timeout_secs: req.timeout_secs,
197            doctor_first: false,
198            idempotency_key: Some(format!("backfill:{hash}:{}", unit.id)),
199            clock: Some(unit.start.to_rfc3339()),
200        };
201        match runner::submit(state.clone(), submit, actor.clone()).await {
202            Ok(resp) => {
203                submitted += 1;
204                reports.push(BackfillUnitRun {
205                    unit: unit.id.clone(),
206                    start: unit.start.to_rfc3339(),
207                    end: unit.end.to_rfc3339(),
208                    status: "submitted".into(),
209                    run_id: Some(resp.run_id),
210                    error: None,
211                });
212            }
213            Err(ServeError::QueueFull { .. }) => {
214                // Deterministic idempotency keys make the whole request
215                // re-POSTable: submitted units replay, the rest submit then.
216                queue_full = true;
217                reports.push(BackfillUnitRun {
218                    unit: unit.id.clone(),
219                    start: unit.start.to_rfc3339(),
220                    end: unit.end.to_rfc3339(),
221                    status: "not_submitted".into(),
222                    run_id: None,
223                    error: Some("run queue full — re-POST the same request to continue".into()),
224                });
225            }
226            Err(other) => return Err(other),
227        }
228    }
229
230    Ok((
231        StatusCode::ACCEPTED,
232        Json(BackfillSubmitResponse {
233            backfill: hash,
234            descriptor,
235            planned,
236            submitted,
237            units: reports,
238        }),
239    ))
240}
241
242/// Rewrite the submitted document for one unit: substitute `${backfill.*}`
243/// tokens across the whole document, namespace the pipeline `name` (so unit
244/// state keys are `{name}-backfill-{unit}::…`, never the live keys), and
245/// force `delivery: at_least_once`. Pure.
246fn rewrite_unit_doc(
247    doc: &Value,
248    unit: &crate::backfill::plan::BackfillUnit,
249    unit_name: &str,
250) -> crate::error::CliResult<String> {
251    let mut d = doc.clone();
252    substitute_unit_tokens(&mut d, unit)?;
253    if let Some(map) = d.as_object_mut() {
254        map.insert("name".into(), Value::String(unit_name.to_string()));
255        map.insert("delivery".into(), Value::String("at_least_once".into()));
256    }
257    serde_yaml::to_string(&d)
258        .map_err(|e| crate::error::CliError::Internal(format!("unit config render: {e}")))
259}
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264    use serde_json::json;
265
266    #[test]
267    fn rewrite_substitutes_namespaces_and_forces_at_least_once() {
268        let utc: chrono_tz::Tz = "UTC".parse().unwrap();
269        let unit = crate::backfill::plan::BackfillUnit {
270            id: "20260601T000000Z".into(),
271            start: parse_boundary("2026-06-01T00:00:00Z", utc).unwrap(),
272            end: parse_boundary("2026-06-02T00:00:00Z", utc).unwrap(),
273        };
274        let doc = json!({
275            "version": 1,
276            "name": "orders",
277            "delivery": "exactly_once",
278            "pipeline": {
279                "source": {"type": "rest", "config": {"url": "https://x/o?s=${backfill.start}&e=${backfill.end}"}},
280                "sink": {"type": "jsonl", "config": {"path": "./out-${backfill.start_date}.jsonl"}}
281            }
282        });
283        let out = rewrite_unit_doc(&doc, &unit, "orders-backfill-20260601T000000Z").unwrap();
284        let back: Value = serde_yaml::from_str(&out).unwrap();
285        assert_eq!(back["name"], "orders-backfill-20260601T000000Z");
286        assert_eq!(back["delivery"], "at_least_once");
287        let url = back["pipeline"]["source"]["config"]["url"]
288            .as_str()
289            .unwrap();
290        assert!(url.contains("s=2026-06-01T00:00:00+00:00"), "{url}");
291        assert!(url.contains("e=2026-06-02T00:00:00+00:00"), "{url}");
292        assert_eq!(
293            back["pipeline"]["sink"]["config"]["path"],
294            "./out-2026-06-01.jsonl"
295        );
296        // The rewritten document is itself a loadable pipeline config.
297        crate::config::parse_with_extension(&out, "yaml").expect("unit doc parses");
298    }
299
300    #[test]
301    fn rewrite_rejects_unknown_token() {
302        let utc: chrono_tz::Tz = "UTC".parse().unwrap();
303        let unit = crate::backfill::plan::BackfillUnit {
304            id: "u".into(),
305            start: parse_boundary("2026-06-01", utc).unwrap(),
306            end: parse_boundary("2026-06-02", utc).unwrap(),
307        };
308        let doc = json!({"pipeline": {"source": {"config": {"q": "${backfill.oops}"}}}});
309        assert!(rewrite_unit_doc(&doc, &unit, "n").is_err());
310    }
311}
faucet_cli/serve/handlers/backfill.rs

faucet_cli/serve/handlers/
backfill.rs
