origin_domain/
sync.rs

1//! Synchronisation bookkeeping.
2//!
3//! The connector decides *how* to fetch. This type records what happened, so the
4//! platform can decide *when* to try again.
5
6use crate::error::ErrorKind;
7use serde::{Deserialize, Serialize};
8use time::OffsetDateTime;
9
10#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
11#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
12#[serde(rename_all = "snake_case", tag = "outcome")]
13pub enum SyncOutcome {
14    /// New data was fetched and stored.
15    Updated,
16    /// The service reported no change (ETag / Last-Modified hit).
17    NotModified,
18    Failed {
19        kind: ErrorKind,
20        message: String,
21    },
22}
23
24/// Why a sync target is being held back beyond its policy cadence.
25///
26/// A service-imposed throttle has different origins but is always handled the same
27/// way by the engine: the next run may not start before a server-chosen instant.
28/// Distinguishing the reason is for logs and the status view.
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
30#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
31#[serde(rename_all = "snake_case")]
32pub enum ThrottleReason {
33    /// Remaining quota or cost the service reported in the *body*, not a header
34    /// (GA4 property quotas, GitHub/Cloudflare GraphQL cost).
35    Quota,
36    /// A minimum poll interval the service named (e.g. GitHub's `X-Poll-Interval`).
37    ServerInterval,
38    /// The service rejected a request as rate-limited and named a retry delay.
39    RateLimited,
40}
41
42#[derive(Debug, Clone, Default, Serialize, Deserialize)]
43#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
44pub struct SyncState {
45    #[serde(with = "time::serde::rfc3339::option")]
46    #[cfg_attr(feature = "ts", ts(type = "string | null"))]
47    pub last_attempt: Option<OffsetDateTime>,
48    #[serde(with = "time::serde::rfc3339::option")]
49    #[cfg_attr(feature = "ts", ts(type = "string | null"))]
50    pub last_success: Option<OffsetDateTime>,
51    pub last_outcome: Option<SyncOutcome>,
52    /// Validators handed back to the service on the next request.
53    pub etag: Option<String>,
54    pub last_modified: Option<String>,
55    /// Consecutive failures, used for exponential backoff.
56    pub failure_streak: u32,
57    /// A server-imposed floor on the next run: the next run may not start before
58    /// this instant, whatever the policy cadence says. Surfaces a quota reset or a
59    /// minimum poll interval and survives restart.
60    #[serde(with = "time::serde::rfc3339::option")]
61    #[cfg_attr(feature = "ts", ts(type = "string | null"))]
62    pub not_before: Option<OffsetDateTime>,
63    /// Why the target is throttled, if it is. For logs and the status view.
64    pub throttle_reason: Option<ThrottleReason>,
65}
66
67impl SyncState {
68    pub fn record(&mut self, at: OffsetDateTime, outcome: SyncOutcome) {
69        self.last_attempt = Some(at);
70        match &outcome {
71            SyncOutcome::Updated | SyncOutcome::NotModified => {
72                self.last_success = Some(at);
73                self.failure_streak = 0;
74            }
75            SyncOutcome::Failed { .. } => {
76                self.failure_streak = self.failure_streak.saturating_add(1);
77            }
78        }
79        self.last_outcome = Some(outcome);
80    }
81
82    pub fn is_failing(&self) -> bool {
83        self.failure_streak > 0
84    }
85
86    /// Set a server-imposed floor on the next run.
87    pub fn throttle_until(&mut self, until: OffsetDateTime, reason: ThrottleReason) {
88        self.not_before = Some(until);
89        self.throttle_reason = Some(reason);
90    }
91
92    /// Clear any server-imposed floor.
93    pub fn clear_throttle(&mut self) {
94        self.not_before = None;
95        self.throttle_reason = None;
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102    use time::macros::datetime;
103
104    #[test]
105    fn a_not_modified_response_still_counts_as_success() {
106        let mut state = SyncState::default();
107        state.record(
108            datetime!(2026-08-23 10:00 UTC),
109            SyncOutcome::Failed {
110                kind: ErrorKind::Network,
111                message: "timeout".into(),
112            },
113        );
114        assert_eq!(state.failure_streak, 1);
115
116        state.record(datetime!(2026-08-23 10:05 UTC), SyncOutcome::NotModified);
117
118        assert_eq!(state.failure_streak, 0);
119        assert_eq!(state.last_success, Some(datetime!(2026-08-23 10:05 UTC)));
120    }
121
122    #[test]
123    fn a_throttle_round_trips_and_can_be_cleared() {
124        let mut state = SyncState::default();
125        let until = datetime!(2026-08-23 10:30 UTC);
126
127        state.throttle_until(until, ThrottleReason::Quota);
128        assert_eq!(state.not_before, Some(until));
129        assert_eq!(state.throttle_reason, Some(ThrottleReason::Quota));
130        assert!(state.not_before.is_some());
131
132        state.clear_throttle();
133        assert_eq!(state.not_before, None);
134        assert_eq!(state.throttle_reason, None);
135    }
136}
origin_domain/sync.rs

origin_domain/
sync.rs
