faucet_source_file/
http.rs

1//! `http(s)://` paths: one remote file, fetched with bounded retries.
2//!
3//! Every request is counted on the pipeline's round-trip recorder (ops `get`
4//! and `head`); a `429` is counted as a throttle, each retry by class, and a
5//! rate-limit sleep is timed, so the run's throttling metrics cover this
6//! source like the API sources.
7
8use faucet_core::FaucetError;
9use faucet_core::io::CallContext;
10use faucet_core::observability::RecorderSlot;
11use faucet_core::resilience::RetryClass;
12use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
13use std::collections::BTreeMap;
14use std::time::Duration;
15
16const BASE_BACKOFF: Duration = Duration::from_millis(100);
17const MAX_BACKOFF: Duration = Duration::from_secs(10);
18
19/// What a `HEAD` reports about the remote file.
20#[derive(Debug, Clone, Default, PartialEq, Eq)]
21pub struct HttpMeta {
22    /// `Last-Modified`, in nanoseconds since the Unix epoch.
23    pub last_modified_ns: Option<i64>,
24}
25
26/// A client for one remote file.
27#[derive(Debug, Clone)]
28pub struct HttpFetcher {
29    client: faucet_core::http::Client,
30    headers: HeaderMap,
31    retries: u32,
32}
33
34impl HttpFetcher {
35    /// Build the client once with the default timeouts (30 s to connect, 300 s
36    /// without a byte); `headers` were validated with the config.
37    pub fn new(headers: &BTreeMap<String, String>, retries: u32) -> Result<Self, FaucetError> {
38        Self::with_timeouts(
39            headers,
40            retries,
41            Duration::from_secs(30),
42            Duration::from_secs(300),
43        )
44    }
45
46    /// Build the client with explicit connect and read (idle) timeouts.
47    pub fn with_timeouts(
48        headers: &BTreeMap<String, String>,
49        retries: u32,
50        connect_timeout: Duration,
51        read_timeout: Duration,
52    ) -> Result<Self, FaucetError> {
53        let mut map = HeaderMap::new();
54        for (k, v) in headers {
55            let name = HeaderName::from_bytes(k.as_bytes())
56                .map_err(|e| FaucetError::Config(format!("file source: header {k:?}: {e}")))?;
57            let value = HeaderValue::from_str(v)
58                .map_err(|e| FaucetError::Config(format!("file source: header {k:?}: {e}")))?;
59            map.insert(name, value);
60        }
61        let mut builder = faucet_core::http::Client::builder()
62            .connect_timeout(connect_timeout)
63            .read_timeout(read_timeout)
64            .max_response_bytes(None);
65        for name in headers.keys() {
66            builder = builder.sensitive_header(name);
67        }
68        let client = builder
69            .build()
70            .map_err(|e| FaucetError::Config(format!("file source: http client: {e}")))?;
71        Ok(Self {
72            client,
73            headers: map,
74            retries,
75        })
76    }
77
78    /// `GET` the file; the response body is left for the caller to stream.
79    pub async fn get(
80        &self,
81        url: &str,
82        slot: &RecorderSlot,
83    ) -> Result<reqwest::Response, FaucetError> {
84        self.get_in(url, slot, &unbounded()).await
85    }
86
87    /// [`get`](Self::get) with every attempt and backoff made through `calls`
88    /// (#880): the context's cancel token and deadline end it.
89    pub async fn get_in(
90        &self,
91        url: &str,
92        slot: &RecorderSlot,
93        calls: &CallContext,
94    ) -> Result<reqwest::Response, FaucetError> {
95        self.send(reqwest::Method::GET, url, "get", slot, calls)
96            .await
97    }
98
99    /// `HEAD` the file for its modification time.
100    pub async fn head(&self, url: &str, slot: &RecorderSlot) -> Result<HttpMeta, FaucetError> {
101        self.head_in(url, slot, &unbounded()).await
102    }
103
104    /// [`head`](Self::head) through `calls` (see [`get_in`](Self::get_in)).
105    pub async fn head_in(
106        &self,
107        url: &str,
108        slot: &RecorderSlot,
109        calls: &CallContext,
110    ) -> Result<HttpMeta, FaucetError> {
111        let resp = self
112            .send(reqwest::Method::HEAD, url, "head", slot, calls)
113            .await?;
114        Ok(HttpMeta {
115            last_modified_ns: resp
116                .headers()
117                .get(reqwest::header::LAST_MODIFIED)
118                .and_then(|v| v.to_str().ok())
119                .and_then(parse_http_date),
120        })
121    }
122
123    async fn send(
124        &self,
125        method: reqwest::Method,
126        url: &str,
127        op: &'static str,
128        slot: &RecorderSlot,
129        calls: &CallContext,
130    ) -> Result<reqwest::Response, FaucetError> {
131        // `slot` counts the round trips; the context bounds them.
132        let calls = calls.clone().with_roundtrip_metering(false);
133        let mut attempt = 0u32;
134        loop {
135            slot.record(op);
136            let result = match calls
137                .call_once(
138                    op,
139                    self.client.send(
140                        self.client
141                            .request(method.clone(), url)
142                            .headers(self.headers.clone()),
143                    ),
144                )
145                .await
146            {
147                Err(spent @ (FaucetError::Cancelled { .. } | FaucetError::Timeout { .. })) => {
148                    return Err(spent);
149                }
150                other => other,
151            };
152            let (class, wait, detail) = match result {
153                Ok(resp) if resp.status().is_success() => return Ok(resp),
154                Ok(resp) if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS => {
155                    slot.throttled();
156                    let wait = retry_after(resp.headers()).unwrap_or_else(|| backoff(attempt));
157                    (
158                        RetryClass::RateLimited,
159                        wait,
160                        format!("HTTP {}", resp.status()),
161                    )
162                }
163                Ok(resp) if resp.status().is_server_error() => (
164                    RetryClass::Http5xx,
165                    backoff(attempt),
166                    format!("HTTP {}", resp.status()),
167                ),
168                Ok(resp) => return Err(faucet_core::http::status_error(resp).await),
169                Err(e) => {
170                    let class = match faucet_core::resilience::classify(&e) {
171                        Some(RetryClass::Timeout) => RetryClass::Timeout,
172                        _ => RetryClass::Connection,
173                    };
174                    let detail = match &e {
175                        FaucetError::Http(inner) => inner.to_string(),
176                        other => other.to_string(),
177                    };
178                    (class, backoff(attempt), detail)
179                }
180            };
181            if attempt >= self.retries {
182                return Err(FaucetError::Source(format!(
183                    "file source: {method} {} failed after {} attempt(s): {}",
184                    faucet_core::util::redact_uri_credentials(url),
185                    attempt + 1,
186                    faucet_core::util::redact_uri_credentials(&detail)
187                )));
188            }
189            slot.retry(class);
190            if class == RetryClass::RateLimited {
191                let _timer = slot.throttle_wait_timer();
192                calls.wait(wait).await?;
193            } else {
194                calls.wait(wait).await?;
195            }
196            attempt += 1;
197        }
198    }
199}
200
201/// The context a fetcher call runs in when the caller gives none: no attempt
202/// bound beyond the client's own timeouts.
203fn unbounded() -> CallContext {
204    CallContext::new().without_timeout()
205}
206
207fn backoff(attempt: u32) -> Duration {
208    BASE_BACKOFF
209        .saturating_mul(1u32 << attempt.min(16))
210        .min(MAX_BACKOFF)
211}
212
213/// `Retry-After` as delta-seconds or an HTTP date, capped at [`MAX_BACKOFF`].
214fn retry_after(headers: &HeaderMap) -> Option<Duration> {
215    let v = headers
216        .get(reqwest::header::RETRY_AFTER)?
217        .to_str()
218        .ok()?
219        .trim();
220    let wait = match v.parse::<u64>() {
221        Ok(secs) => Duration::from_secs(secs),
222        Err(_) => {
223            let at = parse_http_date(v)?;
224            let now = chrono::Utc::now().timestamp_nanos_opt()?;
225            Duration::from_nanos(u64::try_from(at.saturating_sub(now)).unwrap_or(0))
226        }
227    };
228    Some(wait.min(MAX_BACKOFF))
229}
230
231/// An HTTP date (`Sun, 06 Nov 1994 08:49:37 GMT`) in nanoseconds since the
232/// Unix epoch.
233pub fn parse_http_date(v: &str) -> Option<i64> {
234    chrono::DateTime::parse_from_rfc2822(v.trim())
235        .ok()?
236        .timestamp_nanos_opt()
237}
238
239#[cfg(test)]
240mod tests {
241    use super::*;
242
243    #[test]
244    fn backoff_grows_and_caps() {
245        assert_eq!(backoff(0), BASE_BACKOFF);
246        assert_eq!(backoff(1), BASE_BACKOFF * 2);
247        assert_eq!(backoff(40), MAX_BACKOFF);
248    }
249
250    #[test]
251    fn retry_after_reads_seconds_and_dates() {
252        let mut h = HeaderMap::new();
253        assert_eq!(retry_after(&h), None);
254        h.insert(reqwest::header::RETRY_AFTER, HeaderValue::from_static("2"));
255        assert_eq!(retry_after(&h), Some(Duration::from_secs(2)));
256        h.insert(
257            reqwest::header::RETRY_AFTER,
258            HeaderValue::from_static("9999"),
259        );
260        assert_eq!(retry_after(&h), Some(MAX_BACKOFF));
261        h.insert(
262            reqwest::header::RETRY_AFTER,
263            HeaderValue::from_static("Sun, 06 Nov 1994 08:49:37 GMT"),
264        );
265        assert_eq!(
266            retry_after(&h),
267            Some(Duration::ZERO),
268            "a past date waits nothing"
269        );
270        h.insert(
271            reqwest::header::RETRY_AFTER,
272            HeaderValue::from_static("soon"),
273        );
274        assert_eq!(retry_after(&h), None);
275    }
276
277    #[test]
278    fn dates_and_url_names() {
279        assert_eq!(
280            parse_http_date("Thu, 01 Jan 1970 00:00:01 GMT"),
281            Some(1_000_000_000)
282        );
283        assert_eq!(parse_http_date("nope"), None);
284    }
285
286    #[test]
287    fn invalid_headers_are_config_errors() {
288        let mut h = BTreeMap::new();
289        h.insert("bad name".to_string(), "v".to_string());
290        assert!(matches!(
291            HttpFetcher::new(&h, 0),
292            Err(FaucetError::Config(_))
293        ));
294        let mut h = BTreeMap::new();
295        h.insert("x-ok".to_string(), "a\nb".to_string());
296        assert!(matches!(
297            HttpFetcher::new(&h, 0),
298            Err(FaucetError::Config(_))
299        ));
300    }
301}
faucet_source_file/http.rs

faucet_source_file/
http.rs
