surreal_sync_runtime/pipeline/interleaved/
checkpointer.rs

1//! Persisting resumable snapshot progress.
2//!
3//! The watermark loop is decoupled from any concrete storage: it builds a
4//! [`InterleavedSnapshotCheckpoint`] after each chunk and hands it to a
5//! [`SnapshotCheckpointer`]. A no-op implementation is provided for tests, and
6//! [`ManagerCheckpointer`] bridges to the shared [`SyncManager`] /
7//! [`CheckpointStore`] JSON-blob storage.
8
9use anyhow::Result;
10use surreal_sync_core::{CheckpointStore, InterleavedSnapshotCheckpoint, SyncManager, SyncPhase};
11
12pub use surreal_sync_core::SnapshotCheckpointer;
13
14/// A checkpointer that discards progress. Useful for tests and one-shot runs
15/// where resumability is not required.
16#[derive(Debug, Default, Clone, Copy)]
17pub struct NoopCheckpointer;
18
19#[async_trait::async_trait]
20impl SnapshotCheckpointer for NoopCheckpointer {
21    async fn save_progress(&mut self, _checkpoint: &InterleavedSnapshotCheckpoint) -> Result<()> {
22        Ok(())
23    }
24}
25
26/// Persists snapshot checkpoints through the shared [`SyncManager`], reusing
27/// any [`CheckpointStore`] backend (filesystem, SurrealDB, ...).
28pub struct ManagerCheckpointer<S: CheckpointStore> {
29    manager: SyncManager<S>,
30}
31
32impl<S: CheckpointStore> ManagerCheckpointer<S> {
33    /// Wrap a sync manager.
34    pub fn new(manager: SyncManager<S>) -> Self {
35        Self { manager }
36    }
37
38    /// Persist the final checkpoint under the handoff phase, recording the
39    /// position downstream incremental/live processing should resume from.
40    pub async fn save_handoff(&self, checkpoint: &InterleavedSnapshotCheckpoint) -> Result<()> {
41        self.manager
42            .emit_checkpoint(checkpoint, SyncPhase::SnapshotHandoff)
43            .await
44    }
45
46    /// Access the underlying sync manager.
47    pub fn manager(&self) -> &SyncManager<S> {
48        &self.manager
49    }
50}
51
52#[async_trait::async_trait]
53impl<S: CheckpointStore> SnapshotCheckpointer for ManagerCheckpointer<S> {
54    async fn save_progress(&mut self, checkpoint: &InterleavedSnapshotCheckpoint) -> Result<()> {
55        self.manager
56            .emit_checkpoint(checkpoint, SyncPhase::SnapshotProgress)
57            .await
58    }
59}
surreal_sync_runtime/pipeline/interleaved/checkpointer.rs

surreal_sync_runtime/pipeline/interleaved/
checkpointer.rs
