faucet_cli/replication/
spec.rs

1//! Serde config types for the `replication:` block (`faucet replicate`).
2//!
3//! The main `pipeline` is the CDC pipeline (its `source` is a CDC connector,
4//! its `sink` the destination). `replication:` adds the one-time bulk-read
5//! snapshot source used to back-fill before CDC starts. Consumed only by
6//! `faucet replicate`; ignored by `faucet run` (like `schedule:`).
7
8use crate::config::ConnectorSpec;
9use schemars::JsonSchema;
10use serde::{Deserialize, Serialize};
11
12fn default_true() -> bool {
13    true
14}
15
16/// Top-level `replication:` block.
17#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
18#[serde(deny_unknown_fields)]
19pub struct ReplicationSpec {
20    /// Replication strategy. Only `snapshot_then_cdc` is available in v1.
21    pub mode: ReplicationMode,
22    /// One-time bulk-read source used to back-fill the destination before CDC.
23    pub snapshot: SnapshotSpec,
24    /// After the snapshot completes, keep streaming CDC until SIGTERM/SIGINT.
25    /// When `false`, drain CDC once and exit (useful for tests / batch runs).
26    #[serde(default = "default_true")]
27    pub continuous: bool,
28}
29
30/// Replication strategy discriminator.
31#[derive(Debug, Clone, Copy, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
32#[serde(rename_all = "snake_case")]
33pub enum ReplicationMode {
34    /// Capture the CDC position, bulk-snapshot the table, then stream CDC from
35    /// that position. Pair with `write_mode: upsert` for a true mirror.
36    SnapshotThenCdc,
37}
38
39/// The one-time snapshot source (a non-CDC bulk reader of the same upstream DB).
40#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
41#[serde(deny_unknown_fields)]
42pub struct SnapshotSpec {
43    /// Bulk-read source connector (e.g. `postgres` running `SELECT * FROM t`).
44    pub source: ConnectorSpec,
45}
46
47#[cfg(test)]
48mod tests {
49    use super::*;
50
51    #[test]
52    fn parses_minimal_replication_block() {
53        let yaml = r#"
54mode: snapshot_then_cdc
55snapshot:
56  source:
57    type: postgres
58    config: { connection_url: "postgres://x", query: "SELECT * FROM orders" }
59"#;
60        let spec: ReplicationSpec = serde_yaml::from_str(yaml).unwrap();
61        assert_eq!(spec.mode, ReplicationMode::SnapshotThenCdc);
62        assert_eq!(spec.snapshot.source.kind, "postgres");
63        assert!(spec.continuous, "continuous defaults to true");
64    }
65
66    #[test]
67    fn continuous_false_parses() {
68        let yaml = r#"
69mode: snapshot_then_cdc
70continuous: false
71snapshot:
72  source: { type: postgres, config: {} }
73"#;
74        let spec: ReplicationSpec = serde_yaml::from_str(yaml).unwrap();
75        assert!(!spec.continuous);
76    }
77
78    #[test]
79    fn rejects_unknown_field() {
80        let yaml = r#"
81mode: snapshot_then_cdc
82snapshot: { source: { type: postgres, config: {} } }
83bogus: true
84"#;
85        assert!(serde_yaml::from_str::<ReplicationSpec>(yaml).is_err());
86    }
87}
faucet_cli/replication/spec.rs

faucet_cli/replication/
spec.rs
