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.*}`.
12/// Shared with `faucet test` (the `--clock` flag and per-case `clock:` field).
13pub(crate) fn resolve_run_clock(
14    flag: Option<&str>,
15) -> CliResult<chrono::DateTime<chrono::FixedOffset>> {
16    use chrono::{DateTime, NaiveDate, TimeZone, Utc};
17    match flag {
18        None => Ok(Utc::now().fixed_offset()),
19        Some(s) => {
20            if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
21                return Ok(dt);
22            }
23            if let Ok(d) = NaiveDate::parse_from_str(s, "%Y-%m-%d") {
24                let ndt = d.and_hms_opt(0, 0, 0).expect("00:00:00 is valid");
25                return Ok(Utc.from_utc_datetime(&ndt).fixed_offset());
26            }
27            Err(CliError::Config(format!(
28                "--clock '{s}' is not RFC3339 (2026-01-31T00:00:00Z) or a date (2026-01-31)"
29            )))
30        }
31    }
32}
33
34/// Execute the `run` subcommand.
35pub async fn run(args: RunArgs) -> CliResult<()> {
36    let cwd = std::env::current_dir()?;
37    let env_path =
38        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
39    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
40
41    let resolved_config_path: Option<std::path::PathBuf> = if args.from_env {
42        None
43    } else {
44        Some(match args.config.as_ref() {
45            Some(p) => p.clone(),
46            None => {
47                crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?
48            }
49        })
50    };
51
52    let cfg = if args.from_env {
53        if args.profile.is_some() {
54            tracing::warn!(
55                "--profile / FAUCET_PROFILE has no effect in --from-env mode (no config file to compose); ignoring"
56            );
57        }
58        crate::env_config::from_process_env()?
59    } else {
60        PipelineConfig::from_path_async(
61            resolved_config_path
62                .as_ref()
63                .expect("YAML mode always resolves a path above"),
64            args.profile.as_deref(),
65        )
66        .await?
67    };
68
69    #[cfg(not(feature = "cli-tui"))]
70    if args.tui {
71        return Err(CliError::Config(
72            "--tui requires a binary built with the `cli-tui` feature \
73             (e.g. `cargo install faucet-cli --features cli-tui`)"
74                .into(),
75        ));
76    }
77    #[cfg(feature = "cli-tui")]
78    let tui_active = crate::tui::is_tui_session(args.tui);
79    #[cfg(feature = "cli-tui")]
80    let tui_handle = if tui_active {
81        Some(crate::tui::setup_observability(&cfg)?)
82    } else {
83        if args.tui {
84            tracing::info!("--tui: stdout is not a terminal; running without the TUI");
85        }
86        crate::obs::install(&cfg)?;
87        None
88    };
89    #[cfg(not(feature = "cli-tui"))]
90    crate::obs::install(&cfg)?;
91
92    let pipeline_name = cfg.name.clone().unwrap_or_else(|| {
93        resolved_config_path
94            .as_ref()
95            .and_then(|p| p.file_stem())
96            .and_then(|s| s.to_str())
97            .unwrap_or("pipeline")
98            .to_owned()
99    });
100
101    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
102    #[cfg(feature = "lineage")]
103    let lineage = crate::lineage_glue::build_emitter(cfg.lineage.as_ref())
104        .map_err(|e| CliError::Config(format!("lineage: {e}")))?;
105    let resilience = match &cfg.resilience {
106        Some(spec) => Some(spec.to_policy()?),
107        None => None,
108    };
109    #[cfg(feature = "notify")]
110    let notifier = crate::notify::Notifier::from_specs(&cfg.notifications)?;
111    #[cfg(feature = "catalog")]
112    let catalog = match cfg.catalog.as_ref() {
113        Some(spec) => Some(crate::catalog::connect_from_spec(spec).await?),
114        None => None,
115    };
116    let nodes = expand(&cfg)?;
117    // The TUI wires `q` / Ctrl-C to this token: in-flight invocations stop at
118    // their next page boundary and flush (#146 H16). Plain runs keep `None`.
119    #[cfg(feature = "cli-tui")]
120    let tui_cancel = tui_active.then(faucet_core::CancellationToken::new);
121    #[cfg(not(feature = "cli-tui"))]
122    let tui_cancel: Option<faucet_core::CancellationToken> = None;
123    let run_fut = run_expanded(
124        nodes,
125        ExecuteOptions {
126            pipeline_name: pipeline_name.clone(),
127            execution: cfg.execution.clone(),
128            dry_run: args.dry_run,
129            limit: args.limit,
130            state_path_override: args.state_path.clone(),
131            shard: None,
132            auth,
133            clock: resolve_run_clock(args.clock.as_deref())?,
134            // Plain runs have no external cancel signal (the executor still
135            // cooperatively cancels in-flight rows on `on_error: stop`); a
136            // TUI session cancels via `q` / Ctrl-C.
137            cancel: tui_cancel.clone(),
138            resilience,
139            sla: cfg.sla.clone(),
140            #[cfg(feature = "lineage")]
141            lineage,
142            #[cfg(feature = "lineage")]
143            lineage_cfg: cfg.lineage.clone(),
144            #[cfg(feature = "notify")]
145            notifier,
146            #[cfg(feature = "catalog")]
147            catalog,
148        },
149    );
150    #[cfg(feature = "cli-tui")]
151    let summary = match (tui_handle, tui_cancel) {
152        (Some(handle), Some(cancel)) => {
153            let result = crate::tui::drive(run_fut, &pipeline_name, handle, cancel).await;
154            if result
155                .as_ref()
156                .map(|s| s.failure_count() > 0)
157                .unwrap_or(true)
158            {
159                // The failure context lived on the alternate screen — replay
160                // the tail of the log ring to stderr now that it's gone.
161                crate::tui::flush_logs_to_stderr(25);
162            }
163            result?
164        }
165        _ => run_fut.await?,
166    };
167    #[cfg(not(feature = "cli-tui"))]
168    let summary = run_fut.await?;
169
170    let total_written: usize = summary.invocations.iter().map(|i| i.records_written).sum();
171    let success = summary
172        .invocations
173        .iter()
174        .filter(|i| i.error.is_none())
175        .count();
176    let failed = summary.failure_count();
177
178    tracing::info!(
179        pipeline = %pipeline_name,
180        invocations = summary.invocations.len(),
181        succeeded = success,
182        failed,
183        records_written = total_written,
184        "pipeline completed"
185    );
186    println!(
187        "{}: {} invocation{}, {} ok, {} failed, wrote {} record{}",
188        pipeline_name,
189        summary.invocations.len(),
190        if summary.invocations.len() == 1 {
191            ""
192        } else {
193            "s"
194        },
195        success,
196        failed,
197        total_written,
198        if total_written == 1 { "" } else { "s" }
199    );
200
201    // Flush any buffered OTLP telemetry before the process exits (no-op without
202    // the `otel` feature). Done on both the success and failure exit paths.
203    faucet_core::shutdown_otel();
204
205    if summary.had_failures() {
206        return Err(CliError::PipelineHadFailures { count: failed });
207    }
208    Ok(())
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    #[test]
216    fn run_clock_parses_rfc3339_date_and_defaults() {
217        // RFC3339
218        let c = resolve_run_clock(Some("2026-01-31T12:00:00Z")).unwrap();
219        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 12:00");
220        // date-only → midnight UTC
221        let c = resolve_run_clock(Some("2026-01-31")).unwrap();
222        assert_eq!(c.format("%Y-%m-%d %H:%M").to_string(), "2026-01-31 00:00");
223        // default = now (just assert it's Ok / recent year)
224        let c = resolve_run_clock(None).unwrap();
225        assert!(c.format("%Y").to_string().parse::<i32>().unwrap() >= 2025);
226        // bad input errors
227        assert!(resolve_run_clock(Some("not-a-date")).is_err());
228    }
229}
faucet_cli/commands/run.rs

faucet_cli/commands/
run.rs
