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    /// Accepted only so it can be **rejected** with an explanation (#481). One
66    /// `POST /v1/backfill` submits one tracked run per window unit, so a single
67    /// caller-supplied callback has no single run to attach to — firing it N
68    /// times is almost never what the caller means, and silently dropping it
69    /// would leave them waiting forever.
70    #[serde(default)]
71    pub callback: Option<crate::serve::callback::CallbackSpec>,
72}
73
74/// One planned unit's submission outcome.
75#[derive(Debug, Serialize)]
76pub struct BackfillUnitRun {
77    pub unit: String,
78    pub start: String,
79    pub end: String,
80    /// `submitted` | `not_submitted` (queue full — re-POST to continue).
81    pub status: String,
82    #[serde(skip_serializing_if = "Option::is_none")]
83    pub run_id: Option<String>,
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub error: Option<String>,
86}
87
88/// `POST /v1/backfill` success body (202).
89#[derive(Debug, Serialize)]
90pub struct BackfillSubmitResponse {
91    /// Stable range hash — the `backfill` label on every unit run.
92    pub backfill: String,
93    pub descriptor: String,
94    pub planned: usize,
95    pub submitted: usize,
96    pub units: Vec<BackfillUnitRun>,
97}
98
99/// `POST /v1/backfill` → 202 with one tracked run per window unit.
100pub async fn submit_backfill(
101    State(state): State<ServerState>,
102    Extension(actor): Extension<AuthContext>,
103    Json(req): Json<BackfillSubmitRequest>,
104) -> Result<(StatusCode, Json<BackfillSubmitResponse>), ServeError> {
105    // A backfill fans out into one run per window unit, so a single completion
106    // callback is ambiguous. Refuse it explicitly rather than dropping it: a
107    // caller who set it would otherwise wait on a callback that never arrives.
108    if req.callback.is_some() {
109        return Err(ServeError::Unprocessable {
110            message: "`callback` is not supported on /v1/backfill: this submits one run \
111                      per window unit, so there is no single run for a completion callback \
112                      to describe. Poll `GET /v1/runs?labels=backfill:<hash>` for unit \
113                      status, or submit the units individually via `POST /v1/runs` with a \
114                      callback on each"
115                .to_string(),
116            details: None,
117        });
118    }
119
120    // Validate the config loads/expands and gate the window scoping exactly
121    // like the CLI: every root's source must reference a `${backfill.*}` /
122    // `${now.*}` token or each unit would replay identical data.
123    let loaded = load_submission(
124        &req.config,
125        req.config_format.into(),
126        state.default_base().as_ref(),
127    )
128    .await?;
129    let unscoped: Vec<&str> = loaded
130        .nodes
131        .iter()
132        .filter(|n| matches!(n.role, crate::expand::NodeRole::Root))
133        .filter(|n| !has_scoping_tokens(&n.source.config.to_string()))
134        .map(|n| n.id.as_str())
135        .collect();
136    if !unscoped.is_empty() {
137        return Err(ServeError::BadConfig(format!(
138            "root row(s) {} are not scoped to the backfill window — their source configs \
139             reference no `${{backfill.start}}` / `${{backfill.end}}` / `${{now.*}}` token, \
140             so every window would replay identical data (bookmark-positioned backfills \
141             are CLI-only: `faucet backfill --from-bookmark`)",
142            unscoped.join(", ")
143        )));
144    }
145
146    let spec = loaded.cfg.backfill.clone().unwrap_or_default();
147    let tz = match req.timezone.as_deref().or(spec.timezone.as_deref()) {
148        Some(name) => parse_timezone(name).map_err(|e| ServeError::BadConfig(e.to_string()))?,
149        None => chrono_tz::Tz::UTC,
150    };
151    let window = match req.window.as_deref().or(spec.window.as_deref()) {
152        Some(w) => Some(parse_window(w).map_err(|e| ServeError::BadConfig(e.to_string()))?),
153        None => None,
154    };
155    let from = parse_boundary(&req.from, tz).map_err(|e| ServeError::BadConfig(e.to_string()))?;
156    let to = parse_boundary(&req.to, tz).map_err(|e| ServeError::BadConfig(e.to_string()))?;
157    let units = crate::backfill::plan::plan_windows(from, to, window, tz)
158        .map_err(|e| ServeError::BadConfig(e.to_string()))?;
159
160    let base_name = req
161        .name
162        .clone()
163        .or_else(|| loaded.cfg.name.clone())
164        .unwrap_or_else(|| "pipeline".to_string());
165    let descriptor = format!(
166        "time|{}|{}|{}|{base_name}",
167        from.to_rfc3339(),
168        to.to_rfc3339(),
169        window
170            .map(|w| w.to_string())
171            .unwrap_or_else(|| "whole".into()),
172    );
173    let hash = range_hash(&descriptor);
174
175    // Parse the RAW submitted document once; each unit rewrites a copy. The
176    // raw body (not the default-merged config) is submitted so the runner's
177    // own merge/validate path applies per unit.
178    let doc: Value = serde_yaml::from_str(&req.config)
179        .map_err(|e| ServeError::BadConfig(format!("config is not valid YAML/JSON: {e}")))?;
180
181    crate::serve::audit::write(
182        &state,
183        &actor,
184        "backfill.submit",
185        None,
186        Some(hash.clone()),
187        "ok",
188    )
189    .await;
190
191    let planned = units.len();
192    let mut reports = Vec::with_capacity(planned);
193    let mut submitted = 0usize;
194    let mut queue_full = false;
195    for unit in units {
196        if queue_full {
197            reports.push(BackfillUnitRun {
198                unit: unit.id.clone(),
199                start: unit.start.to_rfc3339(),
200                end: unit.end.to_rfc3339(),
201                status: "not_submitted".into(),
202                run_id: None,
203                error: Some("run queue full — re-POST the same request to continue".into()),
204            });
205            continue;
206        }
207        let unit_name = format!("{base_name}-backfill-{}", unit.id);
208        let unit_doc = rewrite_unit_doc(&doc, &unit, &unit_name)
209            .map_err(|e| ServeError::BadConfig(e.to_string()))?;
210        let mut labels = req.labels.clone();
211        labels.insert("backfill".into(), hash.clone());
212        labels.insert("backfill_unit".into(), unit.id.clone());
213        let submit = SubmitRequest {
214            config: unit_doc,
215            config_format: ConfigFormatWire::Yaml,
216            name: Some(unit_name),
217            labels,
218            timeout_secs: req.timeout_secs,
219            doctor_first: false,
220            callback: None,
221            idempotency_key: Some(format!("backfill:{hash}:{}", unit.id)),
222            clock: Some(unit.start.to_rfc3339()),
223        };
224        match runner::submit(state.clone(), submit, actor.clone()).await {
225            Ok(resp) => {
226                submitted += 1;
227                reports.push(BackfillUnitRun {
228                    unit: unit.id.clone(),
229                    start: unit.start.to_rfc3339(),
230                    end: unit.end.to_rfc3339(),
231                    status: "submitted".into(),
232                    run_id: Some(resp.run_id),
233                    error: None,
234                });
235            }
236            Err(ServeError::QueueFull { .. }) => {
237                // Deterministic idempotency keys make the whole request
238                // re-POSTable: submitted units replay, the rest submit then.
239                queue_full = true;
240                reports.push(BackfillUnitRun {
241                    unit: unit.id.clone(),
242                    start: unit.start.to_rfc3339(),
243                    end: unit.end.to_rfc3339(),
244                    status: "not_submitted".into(),
245                    run_id: None,
246                    error: Some("run queue full — re-POST the same request to continue".into()),
247                });
248            }
249            Err(other) => return Err(other),
250        }
251    }
252
253    Ok((
254        StatusCode::ACCEPTED,
255        Json(BackfillSubmitResponse {
256            backfill: hash,
257            descriptor,
258            planned,
259            submitted,
260            units: reports,
261        }),
262    ))
263}
264
265/// Rewrite the submitted document for one unit: substitute `${backfill.*}`
266/// tokens across the whole document, namespace the pipeline `name` (so unit
267/// state keys are `{name}-backfill-{unit}::…`, never the live keys), and
268/// force `delivery: at_least_once`. Pure.
269fn rewrite_unit_doc(
270    doc: &Value,
271    unit: &crate::backfill::plan::BackfillUnit,
272    unit_name: &str,
273) -> crate::error::CliResult<String> {
274    let mut d = doc.clone();
275    substitute_unit_tokens(&mut d, unit)?;
276    if let Some(map) = d.as_object_mut() {
277        map.insert("name".into(), Value::String(unit_name.to_string()));
278        map.insert("delivery".into(), Value::String("at_least_once".into()));
279    }
280    serde_yaml::to_string(&d)
281        .map_err(|e| crate::error::CliError::Internal(format!("unit config render: {e}")))
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287    use serde_json::json;
288
289    #[test]
290    fn rewrite_substitutes_namespaces_and_forces_at_least_once() {
291        let utc: chrono_tz::Tz = "UTC".parse().unwrap();
292        let unit = crate::backfill::plan::BackfillUnit {
293            id: "20260601T000000Z".into(),
294            start: parse_boundary("2026-06-01T00:00:00Z", utc).unwrap(),
295            end: parse_boundary("2026-06-02T00:00:00Z", utc).unwrap(),
296        };
297        let doc = json!({
298            "version": 1,
299            "name": "orders",
300            "delivery": "exactly_once",
301            "pipeline": {
302                "source": {"type": "rest", "config": {"url": "https://x/o?s=${backfill.start}&e=${backfill.end}"}},
303                "sink": {"type": "jsonl", "config": {"path": "./out-${backfill.start_date}.jsonl"}}
304            }
305        });
306        let out = rewrite_unit_doc(&doc, &unit, "orders-backfill-20260601T000000Z").unwrap();
307        let back: Value = serde_yaml::from_str(&out).unwrap();
308        assert_eq!(back["name"], "orders-backfill-20260601T000000Z");
309        assert_eq!(back["delivery"], "at_least_once");
310        let url = back["pipeline"]["source"]["config"]["url"]
311            .as_str()
312            .unwrap();
313        assert!(url.contains("s=2026-06-01T00:00:00+00:00"), "{url}");
314        assert!(url.contains("e=2026-06-02T00:00:00+00:00"), "{url}");
315        assert_eq!(
316            back["pipeline"]["sink"]["config"]["path"],
317            "./out-2026-06-01.jsonl"
318        );
319        // The rewritten document is itself a loadable pipeline config.
320        crate::config::parse_with_extension(&out, "yaml").expect("unit doc parses");
321    }
322
323    #[test]
324    fn rewrite_rejects_unknown_token() {
325        let utc: chrono_tz::Tz = "UTC".parse().unwrap();
326        let unit = crate::backfill::plan::BackfillUnit {
327            id: "u".into(),
328            start: parse_boundary("2026-06-01", utc).unwrap(),
329            end: parse_boundary("2026-06-02", utc).unwrap(),
330        };
331        let doc = json!({"pipeline": {"source": {"config": {"q": "${backfill.oops}"}}}});
332        assert!(rewrite_unit_doc(&doc, &unit, "n").is_err());
333    }
334}
faucet_cli/serve/handlers/backfill.rs

faucet_cli/serve/handlers/
backfill.rs
