syncular_client/
progress.rs

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

syncular_client/
progress.rs
