ocpi_kit/client/
resync.rs

1//! Getting back in sync after a connection loss.
2//!
3//! > *OCPI messages SHOULD NOT be queued. When a client does a POST, PUT or PATCH request and that
4//! > request fails or times out, the client should not queue the message and retry the same
5//! > message again later. When the connection is re-established, it is up to the target-server of
6//! > a connection to GET the current status from the source-server to get back to a synchronized
7//! > state.*
8//!
9//! So the recovery from an outage is not a retry queue — it is a **pull**, and this module builds
10//! the query for it.
11//!
12//! The other half of the advice is about not stampeding:
13//!
14//! > *It is therefore advised to clients pulling lists from a server to do this on a relative low
15//! > polling interval: think in hours, not minutes, and to introduce some splay (randomize the
16//! > length of the poll interface a bit).*
17//!
18//! Spec: 2.3.0 §transport_and_format_offline_behaviour, §transport_and_format_pull_and_push
19
20use core::time::Duration;
21
22use crate::transport::PageQuery;
23use crate::types::DateTime;
24
25/// How far back a resync reaches beyond the last successful pull.
26///
27/// A peer's clock and this one's are not identical, and an object can be written a moment before
28/// its `last_updated` is read, so a resync that starts exactly where the last one ended can miss
29/// an object. Fifteen minutes of overlap costs a handful of duplicate objects, which are
30/// idempotent to apply, and closes the gap.
31pub const DEFAULT_OVERLAP: Duration = Duration::from_mins(15);
32
33/// The interval the specification recommends for routine polling.
34///
35/// > *think in hours, not minutes*
36pub const DEFAULT_POLL_INTERVAL: Duration = Duration::from_hours(4);
37
38/// Builds the pull that brings a receiver back in sync.
39///
40/// ```
41/// use ocpi_kit::client::Resync;
42/// use ocpi_kit::types::DateTime;
43///
44/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
45/// let last_success: DateTime = "2024-03-01T10:00:00Z".parse()?;
46/// let now: DateTime = "2024-03-01T14:00:00Z".parse()?;
47///
48/// let plan = Resync::new().plan(last_success, now);
49/// // The window starts before the last success, so nothing written around the cut is missed.
50/// assert!(plan.query.date_from.unwrap() < last_success);
51/// assert_eq!(plan.query.date_to, Some(now));
52/// # Ok(())
53/// # }
54/// ```
55#[derive(Clone, Copy, Debug)]
56pub struct Resync {
57    overlap: Duration,
58    poll_interval: Duration,
59    splay_fraction: f32,
60    page_limit: Option<u64>,
61}
62
63impl Default for Resync {
64    fn default() -> Self {
65        Self {
66            overlap: DEFAULT_OVERLAP,
67            poll_interval: DEFAULT_POLL_INTERVAL,
68            splay_fraction: 0.2,
69            page_limit: None,
70        }
71    }
72}
73
74impl Resync {
75    /// A resync with the recommended defaults.
76    #[must_use]
77    pub fn new() -> Self {
78        Self::default()
79    }
80
81    /// How far back the window reaches beyond the last successful pull.
82    #[must_use]
83    pub const fn with_overlap(mut self, overlap: Duration) -> Self {
84        self.overlap = overlap;
85        self
86    }
87
88    /// The routine polling interval.
89    #[must_use]
90    pub const fn with_poll_interval(mut self, interval: Duration) -> Self {
91        self.poll_interval = interval;
92        self
93    }
94
95    /// How much of the interval the splay may add or remove, as a fraction. Clamped to `0.0..=1.0`.
96    #[must_use]
97    pub fn with_splay(mut self, fraction: f32) -> Self {
98        self.splay_fraction = fraction.clamp(0.0, 1.0);
99        self
100    }
101
102    /// Asks for a specific page size.
103    #[must_use]
104    pub const fn with_page_limit(mut self, limit: u64) -> Self {
105        self.page_limit = Some(limit);
106        self
107    }
108
109    /// The query that catches up everything changed since `last_success`, up to `now`.
110    #[must_use]
111    pub fn plan(&self, last_success: DateTime, now: DateTime) -> ResyncPlan {
112        let overlap_seconds = i64::try_from(self.overlap.as_secs()).unwrap_or(i64::MAX);
113        let from =
114            DateTime::from_unix_timestamp(last_success.unix_timestamp().saturating_sub(overlap_seconds))
115                .unwrap_or(last_success);
116        let mut query = PageQuery::between(from, now);
117        if let Some(limit) = self.page_limit {
118            query = query.with_limit(limit);
119        }
120        ResyncPlan { query, next_poll_after: self.splayed_interval(now) }
121    }
122
123    /// The query for a routine incremental pull, with no end bound.
124    ///
125    /// An open-ended window keeps picking up objects written while the crawl runs, which is what
126    /// a steady-state poll wants; use [`Resync::plan`] when a closed interval matters.
127    #[must_use]
128    pub fn incremental(&self, last_success: DateTime) -> PageQuery {
129        let overlap_seconds = i64::try_from(self.overlap.as_secs()).unwrap_or(i64::MAX);
130        let from =
131            DateTime::from_unix_timestamp(last_success.unix_timestamp().saturating_sub(overlap_seconds))
132                .unwrap_or(last_success);
133        let mut query = PageQuery::since(from);
134        if let Some(limit) = self.page_limit {
135            query = query.with_limit(limit);
136        }
137        query
138    }
139
140    /// The polling interval with splay applied, so a fleet of clients does not synchronise.
141    ///
142    /// The splay is derived from `seed` rather than from a random number generator, so a given
143    /// client polls at a stable, uncorrelated offset and the behaviour is reproducible in tests.
144    #[must_use]
145    pub fn splayed_interval(&self, seed: DateTime) -> Duration {
146        if self.splay_fraction <= f32::EPSILON {
147            return self.poll_interval;
148        }
149        let span = self.poll_interval.mul_f32(self.splay_fraction);
150        // Map the seed into [0, 2*span) deterministically.
151        let modulus = span.as_secs().saturating_mul(2).max(1);
152        let offset = seed.unix_timestamp().unsigned_abs() % modulus;
153        self.poll_interval.saturating_add(Duration::from_secs(offset)).saturating_sub(span)
154    }
155}
156
157/// What to pull, and when to pull again.
158#[derive(Clone, Debug, PartialEq, Eq)]
159pub struct ResyncPlan {
160    /// The query for the catch-up pull.
161    pub query: PageQuery,
162    /// How long to wait before the next routine poll.
163    pub next_poll_after: Duration,
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    fn dt(s: &str) -> DateTime {
171        s.parse().unwrap()
172    }
173
174    #[test]
175    fn the_window_overlaps_the_last_success() {
176        let plan = Resync::new().plan(dt("2024-03-01T10:00:00Z"), dt("2024-03-01T14:00:00Z"));
177        assert_eq!(plan.query.date_from, Some(dt("2024-03-01T09:45:00Z")));
178        assert_eq!(plan.query.date_to, Some(dt("2024-03-01T14:00:00Z")));
179    }
180
181    #[test]
182    fn an_incremental_poll_has_no_end_bound() {
183        let query =
184            Resync::new().with_overlap(Duration::from_secs(60)).incremental(dt("2024-03-01T10:00:00Z"));
185        assert_eq!(query.date_from, Some(dt("2024-03-01T09:59:00Z")));
186        assert_eq!(query.date_to, None);
187    }
188
189    #[test]
190    fn the_default_interval_is_measured_in_hours_not_minutes() {
191        // "think in hours, not minutes"
192        assert!(DEFAULT_POLL_INTERVAL >= Duration::from_hours(1));
193    }
194
195    #[test]
196    fn splay_stays_within_the_configured_fraction() {
197        let resync = Resync::new().with_splay(0.2);
198        let span = DEFAULT_POLL_INTERVAL.mul_f32(0.2);
199        for seed in
200            ["2024-03-01T14:00:00Z", "2024-03-01T14:00:01Z", "2024-06-17T03:41:59Z", "1970-01-01T00:00:00Z"]
201        {
202            let interval = resync.splayed_interval(dt(seed));
203            assert!(interval >= DEFAULT_POLL_INTERVAL.checked_sub(span).unwrap(), "{seed}: {interval:?}");
204            assert!(interval <= DEFAULT_POLL_INTERVAL + span, "{seed}: {interval:?}");
205        }
206    }
207
208    #[test]
209    fn splay_is_deterministic_and_can_be_switched_off() {
210        let resync = Resync::new();
211        let seed = dt("2024-03-01T14:00:00Z");
212        assert_eq!(resync.splayed_interval(seed), resync.splayed_interval(seed));
213        assert_eq!(Resync::new().with_splay(0.0).splayed_interval(seed), DEFAULT_POLL_INTERVAL);
214    }
215
216    #[test]
217    fn two_clients_starting_at_different_moments_do_not_align() {
218        let resync = Resync::new();
219        let a = resync.splayed_interval(dt("2024-03-01T14:00:00Z"));
220        let b = resync.splayed_interval(dt("2024-03-01T14:07:13Z"));
221        assert_ne!(a, b);
222    }
223}
ocpi_kit/client/resync.rs

ocpi_kit/client/
resync.rs
