faucet_cli/commands/
run.rs

1//! `faucet run` — load a pipeline config, expand the matrix, execute every
2//! invocation under bounded concurrency.
3
4use crate::cli::RunArgs;
5use crate::config::PipelineConfig;
6use crate::error::{CliError, CliResult};
7use crate::executor::{ExecuteOptions, run_expanded};
8use crate::expand::expand;
9
10/// Parse the optional `--clock` override (RFC3339 or `YYYY-MM-DD`), or default
11/// to process start. Returned as a UTC fixed-offset clock for `${now.*}`.
12fn resolve_run_clock(flag: Option<&str>) -> CliResult<chrono::DateTime<chrono::FixedOffset>> {
13    use chrono::{DateTime, NaiveDate, TimeZone, Utc};
14    match flag {
15        None => Ok(Utc::now().fixed_offset()),
16        Some(s) => {
17            if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
18                return Ok(dt);
19            }
20            if let Ok(d) = NaiveDate::parse_from_str(s, "%Y-%m-%d") {
21                let ndt = d.and_hms_opt(0, 0, 0).expect("00:00:00 is valid");
22                return Ok(Utc.from_utc_datetime(&ndt).fixed_offset());
23            }
24            Err(CliError::Config(format!(
25                "--clock '{s}' is not RFC3339 (2026-01-31T00:00:00Z) or a date (2026-01-31)"
26            )))
27        }
28    }
29}
30
31/// Execute the `run` subcommand.
32pub async fn run(args: RunArgs) -> CliResult<()> {
33    let cwd = std::env::current_dir()?;
34    let env_path =
35        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
36    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
37
38    let resolved_config_path: Option<std::path::PathBuf> = if args.from_env {
39        None
40    } else {
41        Some(match args.config.as_ref() {
42            Some(p) => p.clone(),
43            None => {
44                crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?
45            }
46        })
47    };
48
49    let cfg = if args.from_env {
50        if args.profile.is_some() {
51            tracing::warn!(
52                "--profile / FAUCET_PROFILE has no effect in --from-env mode (no config file to compose); ignoring"
53            );
54        }
55        crate::env_config::from_process_env()?
56    } else {
57        PipelineConfig::from_path_async(
58            resolved_config_path
59                .as_ref()
60                .expect("YAML mode always resolves a path above"),
61            args.profile.as_deref(),
62        )
63        .await?
64    };
65
66    crate::obs::install(&cfg)?;
67
68    let pipeline_name = cfg.name.clone().unwrap_or_else(|| {
69        resolved_config_path
70            .as_ref()
71            .and_then(|p| p.file_stem())
72            .and_then(|s| s.to_str())
73            .unwrap_or("pipeline")
74            .to_owned()
75    });
76
77    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
78    #[cfg(feature = "lineage")]
79    let lineage = crate::lineage_glue::build_emitter(cfg.lineage.as_ref())
80        .map_err(|e| CliError::Config(format!("lineage: {e}")))?;
81    let nodes = expand(&cfg)?;
82    let summary = run_expanded(
83        nodes,
84        ExecuteOptions {
85            pipeline_name: pipeline_name.clone(),
86            execution: cfg.execution.clone(),
87            dry_run: args.dry_run,
88            limit: args.limit,
89            state_path_override: args.state_path.clone(),
90            auth,
91            clock: resolve_run_clock(args.clock.as_deref())?,
92            // `faucet run` has no external cancel signal; the executor still
93            // cooperatively cancels in-flight rows on `on_error: stop`.
94            cancel: None,
95            #[cfg(feature = "lineage")]
96            lineage,
97            #[cfg(feature = "lineage")]
98            lineage_cfg: cfg.lineage.clone(),
99        },
100    )
101    .await?;
102
103    let total_written: usize = summary.invocations.iter().map(|i| i.records_written).sum();
104    let success = summary
105        .invocations
106        .iter()
107        .filter(|i| i.error.is_none())
108        .count();
109    let failed = summary.failure_count();
110
111    tracing::info!(
112        pipeline = %pipeline_name,
113        invocations = summary.invocations.len(),
114        succeeded = success,
115        failed,
116        records_written = total_written,
117        "pipeline completed"
118    );
119    println!(
120        "{}: {} invocation{}, {} ok, {} failed, wrote {} record{}",
121        pipeline_name,
122        summary.invocations.len(),
123        if summary.invocations.len() == 1 {
124            ""
125        } else {
126            "s"
127        },
128        success,
129        failed,
130        total_written,
131        if total_written == 1 { "" } else { "s" }
132    );
133
134    if summary.had_failures() {
135        return Err(CliError::PipelineHadFailures { count: failed });
136    }
137    Ok(())
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143
144    #[test]
145    fn run_clock_parses_rfc3339_date_and_defaults() {
146        // RFC3339
147        let c = resolve_run_clock(Some("2026-01-31T12:00:00Z")).unwrap();
148        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 12:00");
149        // date-only → midnight UTC
150        let c = resolve_run_clock(Some("2026-01-31")).unwrap();
151        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 00:00");
152        // default = now (just assert it's Ok / recent year)
153        let c = resolve_run_clock(None).unwrap();
154        assert!(c.format("%Y").to_string().parse::<i32>().unwrap() >= 2025);
155        // bad input errors
156        assert!(resolve_run_clock(Some("not-a-date")).is_err());
157    }
158}
faucet_cli/commands/run.rs

faucet_cli/commands/
run.rs
