faucet_cli/replication/
state.rs

1//! Persistent phase/position marker for `faucet replicate`, plus the pure
2//! phase-decision logic. The marker lives at `{name}::__replication__`; the CDC
3//! bookmark lives at `{name}::cdc` (the CDC node's executor state key).
4
5use crate::error::{CliError, CliResult};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9/// Replication phase recorded in the marker.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "snake_case")]
12pub enum Phase {
13    Snapshot,
14    Cdc,
15}
16
17/// The durable replication marker.
18#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19pub struct ReplicationState {
20    pub phase: Phase,
21    pub snapshot_done: bool,
22    /// CDC start position captured at bootstrap (a CDC bookmark `Value`).
23    pub position: Value,
24}
25
26impl ReplicationState {
27    pub fn to_value(&self) -> CliResult<Value> {
28        serde_json::to_value(self)
29            .map_err(|e| CliError::Internal(format!("replication state serialize: {e}")))
30    }
31    pub fn from_value(v: Value) -> CliResult<Self> {
32        serde_json::from_value(v)
33            .map_err(|e| CliError::Config(format!("replication state parse: {e}")))
34    }
35}
36
37/// State key holding the replication phase marker.
38pub fn marker_key(pipeline_name: &str) -> String {
39    format!("{pipeline_name}::__replication__")
40}
41
42/// State key the CDC node will read/advance (must match the executor's key for
43/// a root node whose id is `cdc`: `{name}::cdc`).
44pub fn cdc_state_key(pipeline_name: &str) -> String {
45    crate::executor::build_state_key(pipeline_name, "cdc", None)
46}
47
48/// What the orchestrator should do given the loaded marker (`None` = fresh).
49#[derive(Debug, PartialEq, Eq)]
50pub enum Plan {
51    /// No marker yet: capture position, seed the CDC bookmark, then snapshot.
52    Bootstrap,
53    /// Marker present, snapshot not yet complete: redo the snapshot (idempotent
54    /// under upsert), then CDC.
55    ResumeSnapshot,
56    /// Marker present, snapshot done: go straight to CDC (resume from bookmark).
57    ResumeCdc,
58}
59
60/// Decide the next action from a loaded marker.
61pub fn plan_from_marker(marker: Option<&ReplicationState>) -> Plan {
62    match marker {
63        None => Plan::Bootstrap,
64        Some(s) if s.snapshot_done => Plan::ResumeCdc,
65        Some(_) => Plan::ResumeSnapshot,
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72    use serde_json::json;
73
74    #[test]
75    fn marker_round_trips() {
76        let s = ReplicationState {
77            phase: Phase::Cdc,
78            snapshot_done: true,
79            position: json!({ "last_lsn": "0/16A4F88" }),
80        };
81        let back = ReplicationState::from_value(s.to_value().unwrap()).unwrap();
82        assert_eq!(back, s);
83    }
84
85    #[test]
86    fn keys_have_expected_shape() {
87        assert_eq!(marker_key("orders"), "orders::__replication__");
88        assert_eq!(cdc_state_key("orders"), "orders::cdc");
89    }
90
91    #[test]
92    fn plan_decisions() {
93        assert_eq!(plan_from_marker(None), Plan::Bootstrap);
94        let snap = ReplicationState {
95            phase: Phase::Snapshot,
96            snapshot_done: false,
97            position: json!(null),
98        };
99        assert_eq!(plan_from_marker(Some(&snap)), Plan::ResumeSnapshot);
100        let done = ReplicationState {
101            phase: Phase::Cdc,
102            snapshot_done: true,
103            position: json!(null),
104        };
105        assert_eq!(plan_from_marker(Some(&done)), Plan::ResumeCdc);
106    }
107
108    #[test]
109    fn marker_key_is_valid_state_key() {
110        faucet_core::state::validate_state_key(&marker_key("orders")).unwrap();
111        faucet_core::state::validate_state_key(&cdc_state_key("orders")).unwrap();
112    }
113}
faucet_cli/replication/state.rs

faucet_cli/replication/
state.rs
