faucet_cli/commands/
replicate.rs

1//! `faucet replicate` — load a config with a `replication:` block, validate it,
2//! and run the two-phase snapshot→CDC orchestration.
3
4use crate::cli::ReplicateArgs;
5use crate::config::PipelineConfig;
6use crate::error::{CliError, CliResult};
7use crate::replication::compiled::CompiledReplication;
8use crate::replication::{ReplicationOptions, run_replication};
9
10/// Execute the `replicate` subcommand.
11pub async fn run(args: ReplicateArgs) -> CliResult<()> {
12    let cwd = std::env::current_dir()?;
13    let env_path =
14        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
15    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
16    let path = match args.config {
17        Some(p) => p,
18        None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
19    };
20
21    let cfg = PipelineConfig::from_path_async(&path, args.profile.as_deref()).await?;
22    let spec = cfg.replication.as_ref().ok_or_else(|| {
23        CliError::Config(
24            "no `replication:` block in config — use `faucet run` for a one-shot run, or add a \
25             `replication:` block (see `faucet schema replication`)"
26                .into(),
27        )
28    })?;
29    // Install observability before compiling the replication spec so the
30    // tracing subscriber is live and `CompiledReplication::compile`'s
31    // non-upsert-sink warning is actually emitted (it would be lost otherwise).
32    crate::obs::install(&cfg)?;
33
34    let compiled = CompiledReplication::compile(spec, &cfg)?;
35
36    let pipeline_name = cfg.name.clone().unwrap_or_else(|| {
37        path.file_stem()
38            .and_then(|s| s.to_str())
39            .unwrap_or("pipeline")
40            .to_owned()
41    });
42    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
43    let resilience = match &cfg.resilience {
44        Some(spec) => Some(spec.to_policy()?),
45        None => None,
46    };
47
48    run_replication(
49        &cfg,
50        &compiled,
51        ReplicationOptions {
52            pipeline_name,
53            execution: cfg.execution.clone(),
54            auth,
55            clock: chrono::Utc::now().fixed_offset(),
56            resilience,
57        },
58    )
59    .await?;
60
61    // Flush any buffered OTLP telemetry before exiting (no-op without `otel`).
62    faucet_core::shutdown_otel();
63
64    println!("replication finished");
65    Ok(())
66}
67
68#[cfg(test)]
69mod tests {
70    use super::*;
71    use std::io::Write;
72
73    /// Write `yaml` to a `faucet.yaml` inside a fresh temp dir and return the
74    /// path (the dir is leaked so the file outlives the call — fine for a test).
75    fn write_config(yaml: &str) -> std::path::PathBuf {
76        let dir = tempfile::tempdir().expect("tempdir");
77        let path = dir.path().join("repl.yaml");
78        let mut f = std::fs::File::create(&path).expect("create config");
79        f.write_all(yaml.as_bytes()).expect("write config");
80        f.flush().expect("flush");
81        // Keep the dir alive for the duration of the test process.
82        std::mem::forget(dir);
83        path
84    }
85
86    fn args(path: std::path::PathBuf) -> ReplicateArgs {
87        ReplicateArgs {
88            config: Some(path),
89            env_file: None,
90            no_env_file: true,
91            profile: None,
92        }
93    }
94
95    /// A valid pipeline config with NO `replication:` block must error out before
96    /// any orchestration runs (no Docker / network reached). Works under default
97    /// features (rest source + jsonl sink are always present).
98    #[tokio::test]
99    async fn errors_when_no_replication_block() {
100        let path = write_config(
101            r#"
102version: 1
103name: plain
104pipeline:
105  source: { type: rest, config: { url: "https://example.com/api" } }
106  sink:   { type: jsonl, config: { path: ./out.jsonl } }
107"#,
108        );
109        let err = run(args(path)).await.unwrap_err();
110        assert!(
111            format!("{err}").contains("replication"),
112            "should mention the missing replication block: {err}"
113        );
114    }
115
116    /// A config WITH a `replication:` block that fails `CompiledReplication::compile`
117    /// (here: `state: memory`, which the durable-state rule rejects) must error
118    /// before `run_replication` — so no Docker is needed. Gated on the connector
119    /// kinds being compiled in so `source_supports_exactly_once` / `source_schema`
120    /// resolve (otherwise compile would fail earlier on an unknown-kind error,
121    /// which is a different, also-acceptable failure but not the branch under test).
122    #[cfg(all(
123        feature = "source-postgres-cdc",
124        feature = "source-postgres",
125        feature = "sink-postgres"
126    ))]
127    #[tokio::test]
128    async fn errors_when_replication_spec_invalid() {
129        let path = write_config(
130            r#"
131version: 1
132name: mirror
133pipeline:
134  source: { type: postgres-cdc, config: { connection_url: "postgres://x", slot_name: s, publication_name: p } }
135  sink:   { type: postgres, config: { connection_url: "postgres://y", table_name: t, column_mapping: auto_map, write_mode: upsert, key: [id] } }
136  state:  { type: memory, config: {} }
137replication:
138  mode: snapshot_then_cdc
139  snapshot:
140    source: { type: postgres, config: { connection_url: "postgres://x", query: "SELECT * FROM t" } }
141"#,
142        );
143        let err = run(args(path)).await.unwrap_err();
144        assert!(
145            format!("{err}").contains("durable state"),
146            "should reject memory state at compile time: {err}"
147        );
148    }
149}
faucet_cli/commands/replicate.rs

faucet_cli/commands/
replicate.rs
