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        crate::env_config::from_process_env()?
51    } else {
52        PipelineConfig::from_path_async(
53            resolved_config_path
54                .as_ref()
55                .expect("YAML mode always resolves a path above"),
56        )
57        .await?
58    };
59
60    crate::obs::install(&cfg)?;
61
62    let pipeline_name = cfg.name.clone().unwrap_or_else(|| {
63        resolved_config_path
64            .as_ref()
65            .and_then(|p| p.file_stem())
66            .and_then(|s| s.to_str())
67            .unwrap_or("pipeline")
68            .to_owned()
69    });
70
71    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
72    let nodes = expand(&cfg)?;
73    let summary = run_expanded(
74        nodes,
75        ExecuteOptions {
76            pipeline_name: pipeline_name.clone(),
77            execution: cfg.execution.clone(),
78            dry_run: args.dry_run,
79            limit: args.limit,
80            state_path_override: args.state_path.clone(),
81            auth,
82            clock: resolve_run_clock(args.clock.as_deref())?,
83            // `faucet run` has no external cancel signal; the executor still
84            // cooperatively cancels in-flight rows on `on_error: stop`.
85            cancel: None,
86        },
87    )
88    .await?;
89
90    let total_written: usize = summary.invocations.iter().map(|i| i.records_written).sum();
91    let success = summary
92        .invocations
93        .iter()
94        .filter(|i| i.error.is_none())
95        .count();
96    let failed = summary.failure_count();
97
98    tracing::info!(
99        pipeline = %pipeline_name,
100        invocations = summary.invocations.len(),
101        succeeded = success,
102        failed,
103        records_written = total_written,
104        "pipeline completed"
105    );
106    println!(
107        "{}: {} invocation{}, {} ok, {} failed, wrote {} record{}",
108        pipeline_name,
109        summary.invocations.len(),
110        if summary.invocations.len() == 1 {
111            ""
112        } else {
113            "s"
114        },
115        success,
116        failed,
117        total_written,
118        if total_written == 1 { "" } else { "s" }
119    );
120
121    if summary.had_failures() {
122        return Err(CliError::PipelineHadFailures { count: failed });
123    }
124    Ok(())
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    #[test]
132    fn run_clock_parses_rfc3339_date_and_defaults() {
133        // RFC3339
134        let c = resolve_run_clock(Some("2026-01-31T12:00:00Z")).unwrap();
135        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 12:00");
136        // date-only → midnight UTC
137        let c = resolve_run_clock(Some("2026-01-31")).unwrap();
138        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 00:00");
139        // default = now (just assert it's Ok / recent year)
140        let c = resolve_run_clock(None).unwrap();
141        assert!(c.format("%Y").to_string().parse::<i32>().unwrap() >= 2025);
142        // bad input errors
143        assert!(resolve_run_clock(Some("not-a-date")).is_err());
144    }
145}
faucet_cli/commands/run.rs

faucet_cli/commands/
run.rs
