feldera_types/transport/
clock.rs

1use std::cmp::max;
2use std::fmt::{self, Display, Formatter};
3use std::str::FromStr;
4
5use chrono::FixedOffset;
6use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
7use utoipa::ToSchema;
8
9/// Fixed timezone offset for the pipeline clock.
10///
11/// Parsed from an ISO-8601 UTC offset string such as `"+05:30"` or
12/// `"-08:00"` and serialized back to the same form.  Wraps
13/// [`chrono::FixedOffset`], which accepts offsets strictly between -24 and
14/// +24 hours.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub struct ClockTimezoneOffset(FixedOffset);
17
18impl ClockTimezoneOffset {
19    /// The offset in milliseconds east of UTC.
20    pub fn offset_ms(&self) -> i64 {
21        i64::from(self.0.local_minus_utc()) * 1_000
22    }
23}
24
25impl FromStr for ClockTimezoneOffset {
26    type Err = String;
27
28    fn from_str(s: &str) -> Result<Self, Self::Err> {
29        FixedOffset::from_str(s).map(Self).map_err(|e| {
30            format!(
31                "invalid timezone offset {s:?} (expected a UTC offset such as \"+05:30\" or \"-08:00\"): {e}"
32            )
33        })
34    }
35}
36
37impl Display for ClockTimezoneOffset {
38    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
39        write!(f, "{}", self.0)
40    }
41}
42
43impl Serialize for ClockTimezoneOffset {
44    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
45        serializer.collect_str(self)
46    }
47}
48
49impl<'de> Deserialize<'de> for ClockTimezoneOffset {
50    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
51        let s = String::deserialize(deserializer)?;
52        s.parse().map_err(de::Error::custom)
53    }
54}
55
56fn is_zero(value: &i64) -> bool {
57    *value == 0
58}
59
60#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize, ToSchema)]
61pub struct ClockConfig {
62    pub clock_resolution_usecs: u64,
63
64    /// Constant offset added to every emitted `NOW()` value, in milliseconds
65    /// east of UTC.  Populated from the `clock_timezone_offset` pipeline
66    /// property; 0 means UTC.
67    #[serde(default, skip_serializing_if = "is_zero")]
68    pub timezone_offset_ms: i64,
69
70    /// Target value for `NOW()` at the worker's first emitted tick, in
71    /// milliseconds since the Unix epoch.
72    ///
73    /// Populated verbatim from `DevTweaks::now_offset` at endpoint
74    /// construction; the wall-clock delta is computed inside the
75    /// connector's worker task from a single `SystemTime::now()`
76    /// reading, so there is no drift between config construction and
77    /// the first emitted tick.  `None` means no shift is applied.
78    #[serde(default, skip_serializing_if = "Option::is_none")]
79    pub now_offset_ms: Option<i64>,
80
81    /// If `true`, the clock does not advance on wall-clock cadence.
82    /// `NOW()` is held at its current value and only advances when an
83    /// external caller invokes the pipeline's `POST /clock/advance`
84    /// endpoint.  Populated from `DevTweaks::now_http_driven`.
85    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
86    pub http_driven: bool,
87}
88
89impl ClockConfig {
90    pub fn clock_resolution_ms(&self) -> u64 {
91        // Refuse to set 0 clock resolution.
92        max((self.clock_resolution_usecs + 500) / 1_000, 1)
93    }
94}
95
96/// Body of `POST /clock/advance`.
97///
98/// `delta_ms` is unsigned; negative values fail JSON deserialization.
99/// `Some(0)` reads the current `NOW()` without moving it or rounding
100/// it; `Some(n)` advances by `n` ms; `None` (`null` or omitted)
101/// advances by one `clock_resolution`.  Non-zero values round up to
102/// the next `clock_resolution` boundary, so a sub-resolution delta
103/// still moves the clock by one full tick.
104#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
105pub struct ClockAdvanceRequest {
106    #[serde(default)]
107    pub delta_ms: Option<u64>,
108}
109
110/// Response of `POST /clock/advance`: the new `NOW()` value as both
111/// milliseconds since epoch (signed; pre-1970 anchors yield negative
112/// values) and an RFC 3339 string.
113#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
114pub struct ClockAdvanceResponse {
115    pub now_ms: i64,
116    pub now: String,
117}
118
119#[cfg(test)]
120mod test {
121    use super::ClockTimezoneOffset;
122
123    #[test]
124    fn timezone_offset_parses_and_round_trips() {
125        const MINUTE_MS: i64 = 60_000;
126        for (input, expected_ms) in [
127            ("+05:30", (5 * 60 + 30) * MINUTE_MS),
128            ("-08:00", -8 * 60 * MINUTE_MS),
129            ("+00:00", 0),
130            ("+14:00", 14 * 60 * MINUTE_MS),
131        ] {
132            let offset: ClockTimezoneOffset =
133                serde_json::from_value(serde_json::json!(input)).unwrap();
134            assert_eq!(offset.offset_ms(), expected_ms, "input {input}");
135            assert_eq!(
136                serde_json::to_value(offset).unwrap(),
137                serde_json::json!(input),
138                "round trip of {input}"
139            );
140        }
141    }
142
143    /// Configurations written before the offset existed must read back
144    /// with no offset.
145    #[test]
146    fn clock_config_without_offset_defaults_to_zero() {
147        let config: super::ClockConfig =
148            serde_json::from_value(serde_json::json!({"clock_resolution_usecs": 1_000_000}))
149                .unwrap();
150        assert_eq!(config.timezone_offset_ms, 0);
151
152        let runtime_config: crate::config::RuntimeConfig =
153            serde_json::from_value(serde_json::json!({"workers": 4})).unwrap();
154        assert_eq!(runtime_config.clock_timezone_offset, None);
155    }
156
157    #[test]
158    fn timezone_offset_rejects_invalid_input() {
159        // Missing sign, garbage, out of chrono's (-24h, +24h) range, and
160        // non-string JSON must all fail deserialization.
161        for input in [
162            serde_json::json!("05:30"),
163            serde_json::json!("banana"),
164            serde_json::json!("+27:00"),
165            serde_json::json!(330),
166        ] {
167            assert!(
168                serde_json::from_value::<ClockTimezoneOffset>(input.clone()).is_err(),
169                "input {input} should be rejected"
170            );
171        }
172    }
173}
feldera_types/transport/clock.rs

feldera_types/transport/
clock.rs
