nodedb_lite/sync/client/
maintenance.rs

1//! Keepalive, backoff, and flow-control-facing accessors.
2
3use std::time::Duration;
4
5use nodedb_types::sync::wire::{PingPongMsg, SyncFrame, SyncMessageType};
6
7use super::config::SyncState;
8use super::state::SyncClient;
9use crate::sync::flow_control::SyncMetricsSnapshot;
10
11impl SyncClient {
12    /// Build a ping frame.
13    pub fn build_ping(&self) -> Option<SyncFrame> {
14        let ping = PingPongMsg {
15            timestamp_ms: crate::runtime::now_millis(),
16            is_pong: false,
17        };
18        SyncFrame::try_encode(SyncMessageType::PingPong, &ping)
19    }
20
21    /// Calculate backoff duration for reconnection attempt N.
22    pub fn backoff_duration(&self, attempt: u32) -> Duration {
23        let base = self.config.min_backoff.as_millis() as u64;
24        let max = self.config.max_backoff.as_millis() as u64;
25        let delay = (base * 2u64.saturating_pow(attempt)).min(max);
26        Duration::from_millis(delay)
27    }
28
29    /// Update pending queue stats in the flow controller.
30    /// Called from the push loop after reading pending deltas.
31    pub async fn update_pending_stats(&self, count: usize, bytes: usize) {
32        let mut flow = self.flow.lock().await;
33        flow.update_pending(count, bytes);
34    }
35
36    /// Check if the pending queue is at capacity (flow control).
37    pub async fn is_queue_full(&self) -> bool {
38        let flow = self.flow.lock().await;
39        flow.is_queue_full()
40    }
41
42    /// Get a snapshot of sync metrics for monitoring/health.
43    pub async fn sync_metrics(&self) -> SyncMetricsSnapshot {
44        let state = *self.state.lock().await;
45        let state_str = match state {
46            SyncState::Disconnected => "disconnected",
47            SyncState::Connecting => "connecting",
48            SyncState::Connected => "connected",
49            SyncState::Reconnecting => "reconnecting",
50        };
51        let flow = self.flow.lock().await;
52        flow.snapshot(state_str, &self.metrics)
53    }
54
55    /// Reset flow controller on reconnect.
56    pub async fn reset_flow_control(&self) {
57        let mut flow = self.flow.lock().await;
58        flow.reset();
59        self.metrics.record_reconnect();
60    }
61}
62
63#[cfg(test)]
64mod tests {
65    use super::*;
66    use crate::sync::client::SyncConfig;
67
68    fn make_config() -> SyncConfig {
69        SyncConfig::new("wss://localhost:9090/sync", "test.jwt.token")
70    }
71
72    #[test]
73    fn backoff_exponential_with_cap() {
74        let client = SyncClient::new(make_config());
75        assert_eq!(client.backoff_duration(0), Duration::from_secs(1));
76        assert_eq!(client.backoff_duration(1), Duration::from_secs(2));
77        assert_eq!(client.backoff_duration(2), Duration::from_secs(4));
78        assert_eq!(client.backoff_duration(3), Duration::from_secs(8));
79        assert_eq!(client.backoff_duration(10), Duration::from_secs(60));
80    }
81
82    #[test]
83    fn ping_frame_is_valid() {
84        let client = SyncClient::new(make_config());
85        let frame = client.build_ping().expect("ping encode");
86        assert_eq!(frame.msg_type, SyncMessageType::PingPong);
87        assert!(!frame.body.is_empty());
88    }
89
90    #[tokio::test]
91    async fn sync_metrics_snapshot() {
92        let client = SyncClient::new(make_config());
93        let snap = client.sync_metrics().await;
94        assert_eq!(snap.state, "disconnected");
95        assert_eq!(snap.pending_count, 0);
96        assert_eq!(snap.deltas_pushed, 0);
97    }
98}
nodedb_lite/sync/client/maintenance.rs

nodedb_lite/sync/client/
maintenance.rs
