syncular_client/
progress.rs

1//! Ephemeral live sync work (SPEC §7.6), observable while the core is busy.
2use serde::{Deserialize, Serialize};
3use std::collections::BTreeMap;
4use std::sync::{Arc, Mutex, Weak};
5
6#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
7#[serde(rename_all = "camelCase")]
8pub struct SyncProgress {
9    pub attempt: u64,
10    pub state: ProgressState,
11    pub phase: ProgressPhase,
12    #[serde(skip_serializing_if = "Option::is_none")]
13    pub subscription_id: Option<String>,
14    #[serde(skip_serializing_if = "Option::is_none")]
15    pub table: Option<String>,
16    #[serde(skip_serializing_if = "Option::is_none")]
17    pub segment_id: Option<String>,
18    pub bytes_received: u64,
19    #[serde(skip_serializing_if = "Option::is_none")]
20    pub bytes_total: Option<u64>,
21    pub rows_processed: u64,
22    #[serde(skip_serializing_if = "Option::is_none")]
23    pub rows_total: Option<u64>,
24    #[serde(skip_serializing_if = "Option::is_none")]
25    pub error_code: Option<String>,
26}
27#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
28#[serde(rename_all = "camelCase")]
29pub enum ProgressState {
30    Running,
31    Complete,
32    Failed,
33}
34#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
35#[serde(rename_all = "camelCase")]
36pub enum ProgressPhase {
37    Request,
38    Download,
39    Import,
40}
41
42pub type SyncProgressListener = Arc<dyn Fn(&SyncProgress) + Send + Sync>;
43
44struct Delivery {
45    callback: SyncProgressListener,
46    revision: Mutex<u64>,
47}
48impl Delivery {
49    fn notify(&self, revision: u64, snapshot: &SyncProgress) {
50        let mut delivered = self.revision.lock().expect("progress delivery lock");
51        if revision <= *delivered {
52            return;
53        }
54        *delivered = revision;
55        // Serialize a late initial replay with updates from the owner thread.
56        // The observer registry is unlocked, so callbacks can read/unsubscribe.
57        let _ =
58            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| (self.callback)(snapshot)));
59    }
60}
61#[derive(Default)]
62struct Inner {
63    snapshot: Option<SyncProgress>,
64    listeners: BTreeMap<u64, Arc<Delivery>>,
65    revision: u64,
66    next_listener: u64,
67}
68
69#[derive(Clone, Default)]
70pub struct ProgressObserver(Arc<Mutex<Inner>>);
71
72/// Keep this guard alive to receive updates; dropping it unsubscribes.
73pub struct ProgressSubscription {
74    inner: Weak<Mutex<Inner>>,
75    id: u64,
76}
77impl Drop for ProgressSubscription {
78    fn drop(&mut self) {
79        if let Some(inner) = self.inner.upgrade() {
80            inner
81                .lock()
82                .expect("progress lock")
83                .listeners
84                .remove(&self.id);
85        }
86    }
87}
88impl ProgressObserver {
89    pub fn snapshot(&self) -> Option<SyncProgress> {
90        self.0.lock().expect("progress lock").snapshot.clone()
91    }
92    pub fn subscribe(
93        &self,
94        listener: impl Fn(&SyncProgress) + Send + Sync + 'static,
95    ) -> ProgressSubscription {
96        let listener = Arc::new(Delivery {
97            callback: Arc::new(listener),
98            revision: Mutex::new(0),
99        });
100        let (id, revision, snapshot) = {
101            let mut inner = self.0.lock().expect("progress lock");
102            inner.next_listener += 1;
103            let id = inner.next_listener;
104            inner.listeners.insert(id, listener.clone());
105            (id, inner.revision, inner.snapshot.clone())
106        };
107        if let Some(snapshot) = snapshot {
108            listener.notify(revision, &snapshot);
109        }
110        ProgressSubscription {
111            inner: Arc::downgrade(&self.0),
112            id,
113        }
114    }
115    pub(crate) fn start(&self) {
116        let attempt = self.snapshot().map_or(1, |s| s.attempt + 1);
117        self.emit(SyncProgress {
118            attempt,
119            state: ProgressState::Running,
120            phase: ProgressPhase::Request,
121            subscription_id: None,
122            table: None,
123            segment_id: None,
124            bytes_received: 0,
125            bytes_total: None,
126            rows_processed: 0,
127            rows_total: None,
128            error_code: None,
129        });
130    }
131    pub(crate) fn update(&self, update: impl FnOnce(&mut SyncProgress)) {
132        if let Some(mut snapshot) = self.snapshot() {
133            if snapshot.state != ProgressState::Running {
134                return;
135            }
136            update(&mut snapshot);
137            self.emit(snapshot);
138        }
139    }
140    fn emit(&self, snapshot: SyncProgress) {
141        let (revision, listeners) = {
142            let mut inner = self.0.lock().expect("progress lock");
143            inner.snapshot = Some(snapshot.clone());
144            inner.revision += 1;
145            (
146                inner.revision,
147                inner.listeners.values().cloned().collect::<Vec<_>>(),
148            )
149        };
150        for listener in listeners {
151            listener.notify(revision, &snapshot);
152        }
153    }
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159    #[test]
160    fn observes_running_work_and_unsubscribes_without_affecting_sync() {
161        let observer = ProgressObserver::default();
162        observer.start();
163        let events = Arc::new(Mutex::new(Vec::new()));
164        let captured = events.clone();
165        let reader = observer.clone();
166        let subscription = observer.subscribe(move |progress| {
167            assert!(reader.snapshot().is_some()); // Callback holds no observer lock.
168            captured.lock().unwrap().push(progress.clone());
169        });
170        let _broken = observer.subscribe(|_| panic!("observer failure"));
171        observer.update(|p| {
172            p.phase = ProgressPhase::Import;
173            p.rows_processed = 1024;
174        });
175        assert_eq!(events.lock().unwrap().len(), 2);
176        drop(subscription);
177        observer.update(|p| {
178            p.state = ProgressState::Failed;
179            p.error_code = Some("sync.transport_failed".into());
180        });
181        assert_eq!(events.lock().unwrap().len(), 2);
182        observer.start();
183        let snapshot = observer.snapshot().unwrap();
184        assert_eq!(snapshot.attempt, 2);
185        assert_eq!(snapshot.rows_processed, 0);
186        assert_eq!(snapshot.error_code, None);
187    }
188    #[test]
189    fn late_subscription_replay_cannot_move_progress_backwards() {
190        let observer = ProgressObserver::default();
191        observer.start();
192        let before = observer.snapshot().unwrap();
193        let events = Arc::new(Mutex::new(Vec::new()));
194        let captured = events.clone();
195        let delivery = Delivery {
196            callback: Arc::new(move |p| captured.lock().unwrap().push(p.rows_processed)),
197            revision: Mutex::new(0),
198        };
199        let mut after = before.clone();
200        after.rows_processed = 1024;
201        delivery.notify(2, &after);
202        delivery.notify(1, &before);
203        assert_eq!(*events.lock().unwrap(), vec![1024]);
204    }
205}
syncular_client/progress.rs

syncular_client/
progress.rs
