surreal-sync-core 0.6.0

Core types, SurrealSink, transform/sink-connect traits, and checkpoint API for surreal-sync
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

//! Sync phase enumeration for checkpoint tracking.

use serde::{Deserialize, Serialize};

/// Represents different phases of the synchronization process.
///
/// Checkpoints are emitted at specific phases during sync operations
/// to enable:
/// - Resumable synchronization (start from last known position)
/// - Incremental sync coordination (use t1/t2 checkpoints)
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SyncPhase {
    /// Checkpoint emitted before full sync begins (t1).
    ///
    /// This checkpoint captures the database state at the moment
    /// before data migration starts. It is used to:
    /// - Resume incremental sync from this point
    /// - Replay changes that occurred during full sync
    FullSyncStart,

    /// Checkpoint emitted after full sync completes (t2).
    ///
    /// This checkpoint captures the database state at the moment
    /// after data migration completes. It marks the point where
    /// incremental sync should stop replaying and switch to live mode.
    FullSyncEnd,

    /// Checkpoint emitted per chunk while a watermark snapshot is in progress.
    ///
    /// This captures the current stream position together with per-table
    /// progress (the last primary key copied and whether the table is done),
    /// enabling a snapshot to resume mid-way after a crash without re-copying
    /// already-copied chunks.
    SnapshotProgress,

    /// Checkpoint emitted when a watermark snapshot completes.
    ///
    /// This records the final stream position the snapshot reached. Downstream
    /// incremental/live processing continues from exactly this position.
    SnapshotHandoff,

    /// Catch-up progress for binlog sync: stream position plus table coverage.
    ///
    /// Updated during incremental sync (position only) and when snapshot batches
    /// complete (position plus covered tables). [`FullSyncEnd`] remains the
    /// immutable t2 boundary written once at snapshot handoff completion.
    CatchUpProgress,
}

impl SyncPhase {
    /// Get the string representation of this phase.
    ///
    /// Used for:
    /// - Checkpoint file naming (e.g., `checkpoint_full_sync_start_2024-01-01.json`)
    /// - Logging and debugging output
    pub fn as_str(&self) -> &str {
        match self {
            SyncPhase::FullSyncStart => "full_sync_start",
            SyncPhase::FullSyncEnd => "full_sync_end",
            SyncPhase::SnapshotProgress => "snapshot_progress",
            SyncPhase::SnapshotHandoff => "snapshot_handoff",
            SyncPhase::CatchUpProgress => "catch_up_progress",
        }
    }
}

impl std::fmt::Display for SyncPhase {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}