faucet_cli/commands/
validate.rs

1//! `faucet validate` — parse + expand a pipeline config without running.
2//!
3//! Surfaces every per-row error with the row id, so a config with multiple
4//! issues reports them together instead of failing at the first one.
5
6use crate::cli::ValidateArgs;
7use crate::config::PipelineConfig;
8use crate::error::{CliError, CliResult};
9use crate::expand::{NodeRole, expand};
10use crate::registry::{sink_schema, source_schema};
11use crate::state::available_state_kinds;
12use crate::transforms::available_transforms;
13
14/// Execute the `validate` subcommand.
15pub async fn run(args: ValidateArgs) -> CliResult<()> {
16    let cwd = std::env::current_dir()?;
17    let env_path =
18        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
19    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
20
21    let path = match args.config {
22        Some(p) => p,
23        None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
24    };
25
26    if args.show_composed {
27        let composed = crate::compose::compose(&path, args.profile.as_deref())?;
28        // Normalize to exactly one trailing newline: the YAML serializer appends
29        // one but `serde_json::to_string_pretty` (JSON-format configs) does not,
30        // and the fast path echoes the file verbatim. A single `\n` keeps
31        // `faucet validate … --show-composed > out.{yaml,json}` well-formed.
32        println!("{}", composed.trim_end_matches('\n'));
33        return Ok(());
34    }
35
36    let cfg = if args.no_secrets {
37        // Grammar / structure only — never touch the network.
38        PipelineConfig::from_path_tolerating_secrets(&path, args.profile.as_deref())?
39    } else {
40        // Real preflight: report each secret reference, then resolve.
41        let refs = crate::secrets::scan_path_refs(&path, args.profile.as_deref())?;
42        let cfg = PipelineConfig::from_path_async(&path, args.profile.as_deref()).await?;
43        for (scheme, reference) in &refs {
44            println!("secret: {scheme}:{reference} → resolved");
45        }
46        cfg
47    };
48    let nodes = expand(&cfg)?;
49
50    // Validate the replication block (snapshot source / CDC source / state) so
51    // `faucet validate` catches misconfiguration without running.
52    if let Some(spec) = &cfg.replication {
53        crate::replication::compiled::CompiledReplication::compile(spec, &cfg)?;
54        println!("replication: mode={:?} — valid", spec.mode);
55    }
56
57    // Validate the schedule block (cron / timezone / bounds) so `faucet validate`
58    // catches schedule misconfiguration in CI without running. Offline-safe.
59    #[cfg(feature = "schedule")]
60    if let Some(spec) = &cfg.schedule {
61        crate::schedule::compiled::CompiledSchedule::compile(spec)?;
62        println!(
63            "schedule: cron '{}' tz '{}' — valid",
64            spec.cron, spec.timezone
65        );
66    }
67
68    // Lineage transport reachability — best-effort. A failure here is only a
69    // warning: lineage emission never blocks a pipeline run.
70    #[cfg(feature = "lineage")]
71    if let Some(lc) = cfg.lineage.as_ref() {
72        match crate::lineage_glue::check_transport(lc).await {
73            Ok(msg) => println!("lineage: {msg}"),
74            Err(msg) => println!("lineage: WARNING — {msg} (lineage never blocks a run)"),
75        }
76    }
77
78    for node in &nodes {
79        // Verifying the schema lookup also catches unknown connector kinds.
80        source_schema(&node.source.kind)?;
81        sink_schema(&node.sink.kind)?;
82        for t in &node.transforms {
83            if !available_transforms().contains(&t.kind.as_str()) {
84                return Err(CliError::UnknownTransform {
85                    name: format!("{} (row '{}')", t.kind, node.id),
86                    available: available_transforms().join(", "),
87                });
88            }
89        }
90        if let Some(state) = &node.state
91            && !available_state_kinds().contains(&state.kind.as_str())
92        {
93            return Err(CliError::UnknownStateStore {
94                name: format!("{} (row '{}')", state.kind, node.id),
95                available: available_state_kinds().join(", "),
96            });
97        }
98    }
99
100    let roots = nodes
101        .iter()
102        .filter(|n| matches!(n.role, NodeRole::Root))
103        .count();
104    let children = nodes.len() - roots;
105    println!(
106        "ok: '{}' rows={} (roots={}, children={}) execution={}",
107        cfg.name.as_deref().unwrap_or("(unnamed)"),
108        nodes.len(),
109        roots,
110        children,
111        cfg.execution
112            .as_ref()
113            .map(|e| format!(
114                "max_concurrent={:?} on_error={:?}",
115                e.max_concurrent.unwrap_or(0),
116                e.on_error
117            ))
118            .unwrap_or_else(|| "(defaults)".to_owned()),
119    );
120    for node in &nodes {
121        let role = match &node.role {
122            NodeRole::Root => "root".to_owned(),
123            NodeRole::Child {
124                parent_id,
125                parent_key,
126            } => {
127                format!("child of '{parent_id}' (parent_key={parent_key})")
128            }
129        };
130        println!(
131            "  - {} [{}] source={} sink={}",
132            node.id, role, node.source.kind, node.sink.kind
133        );
134    }
135    Ok(())
136}
faucet_cli/commands/validate.rs

faucet_cli/commands/
validate.rs
