surreal_sync_core/checkpoint/
snapshot.rs

1//! Resumable checkpoint for the watermark-based interleaved snapshot strategy.
2//!
3//! Unlike the scalar full-sync checkpoints (which carry a single stream
4//! position), an interleaved snapshot copies tables in primary-key-ordered
5//! chunks while concurrently consuming the change stream. To resume after a
6//! crash it must remember both the current stream position and how far each
7//! table has been copied.
8//!
9//! Both the stream position and the per-table last primary key are stored as
10//! opaque JSON values so this type stays independent of any particular source
11//! backend (an LSN string, an integer sequence id, a single- or composite-key
12//! tuple, etc. all serialize to JSON).
13
14use serde::{Deserialize, Serialize};
15
16/// Progress for a single table within an interleaved snapshot.
17#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
18pub struct SnapshotTableProgress {
19    /// Table name.
20    pub name: String,
21    /// Last primary key copied for this table, serialized as JSON.
22    ///
23    /// `None` means no chunk has been copied yet; resume starts from the
24    /// beginning of the table.
25    pub last_pk: Option<serde_json::Value>,
26    /// Whether the table has been fully copied.
27    pub done: bool,
28}
29
30/// Resumable checkpoint describing the state of an in-progress interleaved
31/// snapshot.
32#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
33pub struct InterleavedSnapshotCheckpoint {
34    /// Current reconciliation stream position, serialized as JSON.
35    ///
36    /// On resume the snapshot continues consuming the change stream from this
37    /// position; on completion this is the position handed off to downstream
38    /// replication-tail processing.
39    #[serde(alias = "stream_pos")]
40    pub reconciliation_pos: serde_json::Value,
41    /// Per-table copy progress.
42    pub tables: Vec<SnapshotTableProgress>,
43}
44
45impl InterleavedSnapshotCheckpoint {
46    /// Create a new snapshot checkpoint.
47    pub fn new(reconciliation_pos: serde_json::Value, tables: Vec<SnapshotTableProgress>) -> Self {
48        Self {
49            reconciliation_pos,
50            tables,
51        }
52    }
53
54    /// Whether every table has been fully copied.
55    pub fn all_done(&self) -> bool {
56        self.tables.iter().all(|t| t.done)
57    }
58}
59
60impl crate::checkpoint::Checkpoint for InterleavedSnapshotCheckpoint {
61    const DATABASE_TYPE: &'static str = "interleaved_snapshot";
62
63    fn to_cli_string(&self) -> String {
64        serde_json::to_string(self).unwrap_or_default()
65    }
66
67    fn from_cli_string(s: &str) -> anyhow::Result<Self> {
68        Ok(serde_json::from_str(s)?)
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use super::*;
75    use crate::checkpoint::{Checkpoint, CheckpointFile, SyncPhase};
76
77    fn sample() -> InterleavedSnapshotCheckpoint {
78        InterleavedSnapshotCheckpoint::new(
79            serde_json::json!("0/16B3748"),
80            vec![
81                SnapshotTableProgress {
82                    name: "users".to_string(),
83                    last_pk: Some(serde_json::json!([{ "type": "Int64", "value": 42 }])),
84                    done: false,
85                },
86                SnapshotTableProgress {
87                    name: "orders".to_string(),
88                    last_pk: None,
89                    done: true,
90                },
91            ],
92        )
93    }
94
95    #[test]
96    fn cli_string_roundtrip() {
97        let original = sample();
98        let s = original.to_cli_string();
99        let decoded = InterleavedSnapshotCheckpoint::from_cli_string(&s).unwrap();
100        assert_eq!(original, decoded);
101    }
102
103    #[test]
104    fn checkpoint_file_roundtrip() {
105        let original = sample();
106        let file = CheckpointFile::new(&original, SyncPhase::SnapshotProgress).unwrap();
107        assert_eq!(
108            file.database_type(),
109            InterleavedSnapshotCheckpoint::DATABASE_TYPE
110        );
111        let decoded: InterleavedSnapshotCheckpoint = file.parse().unwrap();
112        assert_eq!(original, decoded);
113    }
114
115    #[test]
116    fn all_done_reflects_table_state() {
117        let mut cp = sample();
118        assert!(!cp.all_done());
119        for t in &mut cp.tables {
120            t.done = true;
121        }
122        assert!(cp.all_done());
123    }
124}
surreal_sync_core/checkpoint/snapshot.rs

surreal_sync_core/checkpoint/
snapshot.rs
