launchdarkly_server_sdk/fdv2/
source.rs

1use std::time::Duration;
2
3use rand::Rng;
4
5use crate::stores::change_set::ChangeSet;
6
7use super::model::Selector;
8
9const FALLBACK_HEADER: &str = "X-LD-FD-Fallback";
10const FALLBACK_TTL_HEADER: &str = "X-LD-FD-Fallback-TTL";
11const DEFAULT_FALLBACK_TTL: Duration = Duration::from_secs(60 * 60);
12/// The longest server-supplied fallback TTL that is honored; longer values fall back to the default.
13const MAX_FALLBACK_TTL: Duration = Duration::from_secs(60 * 60);
14
15/// Classifies why a data source is interrupted or has failed.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub enum ErrorKind {
18    /// The cause is not one of the more specific kinds.
19    Unknown,
20    /// The request failed at the network layer.
21    NetworkError,
22    /// The server returned an error status code.
23    ErrorResponse {
24        /// The HTTP status code.
25        status_code: u16,
26    },
27    /// The response could not be parsed.
28    InvalidData,
29}
30
31/// Describes an error surfaced by a data source.
32#[derive(Debug, Clone)]
33pub struct ErrorInfo {
34    /// The kind of error.
35    pub kind: ErrorKind,
36    /// A human-readable description.
37    pub message: String,
38}
39
40/// An instruction from LaunchDarkly to fall back to the FDv1 protocol.
41#[derive(Debug, Clone)]
42pub struct FDv1FallbackDirective {
43    /// How long to stay on FDv1 before retrying FDv2.
44    pub ttl: Duration,
45}
46
47impl FDv1FallbackDirective {
48    /// Builds a directive from a server-supplied TTL. A value in the `(0, 1 hour]`
49    /// range is honored as-is; anything else -- absent, zero, or longer than an
50    /// hour -- uses a jittered one-hour default.
51    pub(super) fn from_ttl(ttl: Option<Duration>) -> Self {
52        let ttl = match ttl {
53            Some(t) if t > Duration::ZERO && t <= MAX_FALLBACK_TTL => t,
54            _ => jittered_default_ttl(),
55        };
56        Self { ttl }
57    }
58}
59
60/// Jitters the default TTL down by up to half so a fleet that all hits the default
61/// doesn't retry FDv2 in lockstep. Server-supplied TTLs are jittered upstream and
62/// are left untouched.
63fn jittered_default_ttl() -> Duration {
64    DEFAULT_FALLBACK_TTL.mul_f64(rand::rng().random_range(0.5..=1.0))
65}
66
67pub(super) fn read_fallback_directive<'a>(
68    lookup: impl Fn(&str) -> Option<&'a str>,
69) -> Option<FDv1FallbackDirective> {
70    let flag = lookup(FALLBACK_HEADER)?;
71    if !flag.eq_ignore_ascii_case("true") {
72        return None;
73    }
74    let ttl = lookup(FALLBACK_TTL_HEADER)
75        .and_then(|s| s.parse::<u64>().ok())
76        .map(Duration::from_secs);
77    Some(FDv1FallbackDirective::from_ttl(ttl))
78}
79
80/// The outcome of a single data source poll or stream read.
81#[derive(Debug)]
82pub enum FDv2SourceResult {
83    /// A set of flag and segment changes.
84    ChangeSet(ChangeSet),
85    /// A transient failure; the source may recover.
86    Interrupted(ErrorInfo),
87    /// An unrecoverable failure; the source is done.
88    TerminalError(ErrorInfo),
89    /// The server asked the source to disconnect.
90    Goodbye,
91}
92
93/// A source result paired with any FDv1 fallback directive seen on the same response.
94#[derive(Debug)]
95pub struct FDv2SourceEvent {
96    /// The source result.
97    pub result: FDv2SourceResult,
98    /// Present when the response carried an FDv1 fallback directive.
99    pub fdv1_fallback: Option<FDv1FallbackDirective>,
100}
101
102/// The future returned by an initializer or synchronizer as it produces an event.
103pub type FDv2SourceEventFuture<'a> =
104    std::pin::Pin<Box<dyn std::future::Future<Output = FDv2SourceEvent> + Send + 'a>>;
105
106/// A data source that can obtain an initial payload.
107pub trait Initializer: Send {
108    /// Runs once to obtain an initial payload.
109    fn run(&mut self) -> FDv2SourceEventFuture<'_>;
110    /// The name used in logs.
111    fn name(&self) -> &str;
112}
113
114/// A data source that keeps flag data up to date.
115pub trait Synchronizer: Send {
116    /// Fetches the next batch of changes after the given selector.
117    fn next(&mut self, selector: Selector) -> FDv2SourceEventFuture<'_>;
118    /// The name used in logs.
119    fn name(&self) -> &str;
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125    use std::collections::HashMap;
126
127    fn headers<'a>(pairs: &'a [(&'a str, &'a str)]) -> impl Fn(&str) -> Option<&'a str> + 'a {
128        let map: HashMap<&'a str, &'a str> = pairs.iter().copied().collect();
129        move |k| map.get(k).copied()
130    }
131
132    // The default is jittered down by up to half, so it lands in [30 min, 1 hour].
133    fn assert_is_jittered_default(ttl: Duration) {
134        assert!(ttl <= DEFAULT_FALLBACK_TTL && ttl >= DEFAULT_FALLBACK_TTL / 2);
135    }
136
137    #[test]
138    fn fallback_absent_header_returns_none() {
139        assert!(read_fallback_directive(headers(&[])).is_none());
140    }
141
142    #[test]
143    fn fallback_header_value_other_than_true_returns_none() {
144        assert!(read_fallback_directive(headers(&[("X-LD-FD-Fallback", "false")])).is_none());
145    }
146
147    #[test]
148    fn fallback_header_uppercase_true_uses_default_ttl() {
149        let d =
150            read_fallback_directive(headers(&[("X-LD-FD-Fallback", "TRUE")])).expect("directive");
151        assert_is_jittered_default(d.ttl);
152    }
153
154    #[test]
155    fn fallback_ttl_header_is_parsed() {
156        let d = read_fallback_directive(headers(&[
157            ("X-LD-FD-Fallback", "true"),
158            ("X-LD-FD-Fallback-TTL", "60"),
159        ]))
160        .expect("directive");
161        assert_eq!(d.ttl, Duration::from_secs(60));
162    }
163
164    #[test]
165    fn fallback_ttl_header_malformed_uses_default() {
166        let d = read_fallback_directive(headers(&[
167            ("X-LD-FD-Fallback", "true"),
168            ("X-LD-FD-Fallback-TTL", "not-a-number"),
169        ]))
170        .expect("directive");
171        assert_is_jittered_default(d.ttl);
172    }
173
174    #[test]
175    fn fallback_ttl_zero_uses_default() {
176        let d = read_fallback_directive(headers(&[
177            ("X-LD-FD-Fallback", "true"),
178            ("X-LD-FD-Fallback-TTL", "0"),
179        ]))
180        .expect("directive");
181        assert_is_jittered_default(d.ttl);
182    }
183
184    #[test]
185    fn fallback_ttl_longer_than_an_hour_uses_default() {
186        let d = read_fallback_directive(headers(&[
187            ("X-LD-FD-Fallback", "true"),
188            ("X-LD-FD-Fallback-TTL", "3601"),
189        ]))
190        .expect("directive");
191        assert_is_jittered_default(d.ttl);
192    }
193}
launchdarkly_server_sdk/fdv2/source.rs

launchdarkly_server_sdk/fdv2/
source.rs
