tango/
transport.rs

1//! Request loop, retry policy, header parsing, and validation-message extraction.
2//!
3//! This module is the core of the SDK. Public surface ([`RateLimitInfo`]) is
4//! re-exported from `crate`. Internals are crate-private — resource methods
5//! call [`Client::request_json`](crate::Client::request_json) and friends, not
6//! this module directly.
7
8use crate::error::{Error, ErrorBody, Result};
9use reqwest::{header::HeaderMap, Method, RequestBuilder, Response, StatusCode};
10use serde::de::DeserializeOwned;
11use serde_json::Value;
12use std::time::Duration;
13
14pub(crate) const MAX_BACKOFF: Duration = Duration::from_secs(10);
15
16pub(crate) const API_KEY_HEADER: &str = "X-API-KEY";
17pub(crate) const RATE_LIMIT_REMAINING_HDR: &str = "x-ratelimit-remaining";
18pub(crate) const RATE_LIMIT_LIMIT_HDR: &str = "x-ratelimit-limit";
19pub(crate) const RATE_LIMIT_RESET_HDR: &str = "x-ratelimit-reset";
20pub(crate) const RATE_LIMIT_TYPE_HDR: &str = "x-ratelimit-type";
21pub(crate) const RETRY_AFTER_HDR: &str = "retry-after";
22
23/// A snapshot of the rate-limit headers from the most recent response.
24///
25/// Returned by [`Client::rate_limit_info`](crate::Client::rate_limit_info).
26/// Any field set to `None` was either absent from the response or could not
27/// be parsed as an integer.
28#[derive(Debug, Clone, Default, PartialEq, Eq)]
29pub struct RateLimitInfo {
30    /// `X-RateLimit-Remaining` — calls remaining in the current bucket.
31    pub remaining: Option<i64>,
32    /// `X-RateLimit-Limit` — the maximum calls in the current bucket.
33    pub limit: Option<i64>,
34    /// `X-RateLimit-Reset` — seconds until the bucket refills.
35    pub reset_in: Option<i64>,
36    /// `Retry-After` — server-suggested wait before retrying, in seconds.
37    pub retry_after: Option<i64>,
38    /// `X-RateLimit-Type` — bucket discriminator (e.g. `"minute"`, `"hour"`).
39    pub limit_type: Option<String>,
40}
41
42impl RateLimitInfo {
43    pub(crate) fn from_headers(h: &HeaderMap) -> Self {
44        Self {
45            remaining: parse_int_header(h, RATE_LIMIT_REMAINING_HDR),
46            limit: parse_int_header(h, RATE_LIMIT_LIMIT_HDR),
47            reset_in: parse_int_header(h, RATE_LIMIT_RESET_HDR),
48            retry_after: parse_int_header(h, RETRY_AFTER_HDR),
49            limit_type: h
50                .get(RATE_LIMIT_TYPE_HDR)
51                .and_then(|v| v.to_str().ok())
52                .map(str::to_string),
53        }
54    }
55}
56
57fn parse_int_header(h: &HeaderMap, key: &str) -> Option<i64> {
58    h.get(key)
59        .and_then(|v| v.to_str().ok())
60        .map(str::trim)
61        .and_then(|s| s.parse::<i64>().ok())
62}
63
64/// Parse the `Retry-After` header into a duration. Accepts both delta-seconds
65/// and HTTP-date forms; returns `Duration::ZERO` on parse failure. The value
66/// is capped at [`MAX_BACKOFF`].
67pub(crate) fn parse_retry_after(h: &HeaderMap) -> Duration {
68    let Some(raw) = h.get(RETRY_AFTER_HDR).and_then(|v| v.to_str().ok()) else {
69        return Duration::ZERO;
70    };
71    let raw = raw.trim();
72    if let Ok(secs) = raw.parse::<u64>() {
73        return Duration::from_secs(secs).min(MAX_BACKOFF);
74    }
75    // HTTP-date form. We don't pull in chrono; httpdate would be a dep we want
76    // to avoid. The Tango server emits delta-seconds in practice, so we accept
77    // a parse failure here and let the exponential backoff take over.
78    Duration::ZERO
79}
80
81/// Extract a human-readable validation message from a 400 response body.
82///
83/// Mirrors the Go SDK's `extractValidationMessage`. Tries the envelope keys
84/// `detail`, `message`, `error` in priority order; falls back to sorted-key
85/// iteration over the remaining keys (preferring array values over strings),
86/// so the surfaced message is deterministic across runs.
87pub(crate) fn extract_validation_message(body: Option<&Value>) -> String {
88    const FALLBACK: &str = "invalid request parameters";
89    let Some(Value::Object(map)) = body else {
90        return FALLBACK.to_string();
91    };
92    for key in ["detail", "message", "error"] {
93        if let Some(Value::String(s)) = map.get(key) {
94            if !s.is_empty() {
95                return format!("invalid request parameters: {s}");
96            }
97        }
98    }
99    // Field-error shape — walk keys in sorted order for determinism.
100    let mut keys: Vec<&String> = map.keys().collect();
101    keys.sort();
102    for k in keys {
103        match map.get(k) {
104            Some(Value::Array(arr)) if !arr.is_empty() => {
105                if let Some(Value::String(s)) = arr.first() {
106                    if !s.is_empty() {
107                        return format!("invalid request parameters: {s}");
108                    }
109                }
110            }
111            Some(Value::String(s)) if !s.is_empty() => {
112                return format!("invalid request parameters: {s}");
113            }
114            _ => {}
115        }
116    }
117    FALLBACK.to_string()
118}
119
120/// Body the SDK sends with a request. Used by [`Client::request_json`] and
121/// [`Client::request_bytes`] callers.
122#[derive(Debug)]
123pub(crate) enum Body<'a> {
124    None,
125    Json(&'a Value),
126}
127
128pub(crate) async fn send_with_retries(
129    inner: &crate::client::ClientInner,
130    method: Method,
131    url: reqwest::Url,
132    body: Body<'_>,
133) -> Result<Vec<u8>> {
134    send_with_retries_and_headers(inner, method, url, body)
135        .await
136        .map(|(bytes, _)| bytes)
137}
138
139/// Like [`send_with_retries`], but also return the successful response's headers.
140pub(crate) async fn send_with_retries_and_headers(
141    inner: &crate::client::ClientInner,
142    method: Method,
143    url: reqwest::Url,
144    body: Body<'_>,
145) -> Result<(Vec<u8>, HeaderMap)> {
146    // Pre-serialize the JSON body once; we re-use the bytes across attempts.
147    let body_bytes = match body {
148        Body::None => None,
149        Body::Json(v) => Some(serde_json::to_vec(v)?),
150    };
151    retry(inner, || {
152        attempt_once(inner, method.clone(), url.clone(), body_bytes.as_deref())
153    })
154    .await
155}
156
157/// GET `url` on the no-redirect client and return the resolved `Location` of
158/// its 3xx response, under the same retry policy as every other request.
159pub(crate) async fn redirect_location_with_retries(
160    inner: &crate::client::ClientInner,
161    url: reqwest::Url,
162) -> Result<String> {
163    retry(inner, || attempt_redirect(inner, url.clone())).await
164}
165
166async fn retry<T, F, Fut>(inner: &crate::client::ClientInner, mut op: F) -> Result<T>
167where
168    F: FnMut() -> Fut,
169    Fut: std::future::Future<Output = Result<T>>,
170{
171    let max_attempts = inner.retries.saturating_add(1);
172    let mut attempt: u32 = 0;
173
174    loop {
175        let err = match op().await {
176            Ok(v) => return Ok(v),
177            Err(e) => e,
178        };
179
180        if !err.is_retryable() || attempt + 1 >= max_attempts {
181            return Err(err);
182        }
183
184        let wait = if let Error::RateLimit { retry_after, .. } = &err {
185            let r = u64::from(*retry_after);
186            if r > 0 {
187                Duration::from_secs(r).min(MAX_BACKOFF)
188            } else {
189                backoff_for(inner.retry_backoff, attempt)
190            }
191        } else {
192            backoff_for(inner.retry_backoff, attempt)
193        };
194
195        tokio::time::sleep(wait).await;
196        attempt += 1;
197    }
198}
199
200fn backoff_for(base: Duration, attempt: u32) -> Duration {
201    let mult = 1u32 << attempt.min(6); // cap shift at 6 to avoid overflow
202    base.saturating_mul(mult).min(MAX_BACKOFF)
203}
204
205fn build_request(
206    inner: &crate::client::ClientInner,
207    http: &reqwest::Client,
208    method: Method,
209    url: reqwest::Url,
210    body_bytes: Option<&[u8]>,
211) -> RequestBuilder {
212    let mut req: RequestBuilder = http.request(method, url);
213    req = req.header(reqwest::header::ACCEPT, "application/json");
214    if !inner.api_key.is_empty() {
215        req = req.header(API_KEY_HEADER, &inner.api_key);
216    }
217    if !inner.user_agent.is_empty() {
218        req = req.header(reqwest::header::USER_AGENT, &inner.user_agent);
219    }
220    if let Some(bytes) = body_bytes {
221        req = req
222            .header(reqwest::header::CONTENT_TYPE, "application/json")
223            .body(bytes.to_vec());
224    }
225    if !inner.timeout.is_zero() {
226        req = req.timeout(inner.timeout);
227    }
228    req
229}
230
231/// Send `req` and return its status, headers and body, recording the headers
232/// for [`Client::rate_limit_info`](crate::Client::rate_limit_info).
233async fn send_raw(
234    inner: &crate::client::ClientInner,
235    req: RequestBuilder,
236) -> Result<(StatusCode, HeaderMap, Vec<u8>)> {
237    let resp: Response = match req.send().await {
238        Ok(r) => r,
239        Err(e) => {
240            if e.is_timeout() {
241                return Err(Error::Timeout {
242                    timeout: inner.timeout,
243                });
244            }
245            return Err(Error::Transport(e));
246        }
247    };
248
249    let status = resp.status();
250    let headers = resp.headers().clone();
251    // Snapshot rate-limit/headers for observability.
252    inner.set_last_response(&headers);
253
254    let bytes = match resp.bytes().await {
255        Ok(b) => b.to_vec(),
256        Err(e) => return Err(Error::Transport(e)),
257    };
258    Ok((status, headers, bytes))
259}
260
261async fn attempt_once(
262    inner: &crate::client::ClientInner,
263    method: Method,
264    url: reqwest::Url,
265    body_bytes: Option<&[u8]>,
266) -> Result<(Vec<u8>, HeaderMap)> {
267    let req = build_request(inner, &inner.http, method, url, body_bytes);
268    let (status, headers, bytes) = send_raw(inner, req).await?;
269    if status.is_success() {
270        return Ok((bytes, headers));
271    }
272    Err(decode_error(status, &headers, &bytes))
273}
274
275async fn attempt_redirect(inner: &crate::client::ClientInner, url: reqwest::Url) -> Result<String> {
276    let req = build_request(
277        inner,
278        &inner.http_no_redirect,
279        Method::GET,
280        url.clone(),
281        None,
282    );
283    let (status, headers, bytes) = send_raw(inner, req).await?;
284    if status.is_redirection() {
285        let location = headers
286            .get(reqwest::header::LOCATION)
287            .and_then(|v| v.to_str().ok())
288            .filter(|s| !s.is_empty())
289            .ok_or_else(|| Error::Api {
290                status: status.as_u16(),
291                message: "redirect response carried no Location header".into(),
292                response: None,
293            })?;
294        return url
295            .join(location)
296            .map(String::from)
297            .map_err(|e| Error::Build(format!("parse redirect location {location}: {e}")));
298    }
299    if status.is_success() {
300        return Err(Error::Api {
301            status: status.as_u16(),
302            message: format!("expected a redirect, got status {}", status.as_u16()),
303            response: None,
304        });
305    }
306    Err(decode_error(status, &headers, &bytes))
307}
308
309fn decode_error(status: StatusCode, headers: &HeaderMap, body: &[u8]) -> Error {
310    let parsed_value: Option<Value> = if body.is_empty() {
311        None
312    } else {
313        serde_json::from_slice(body).ok()
314    };
315    let body_message = parsed_value.as_ref().and_then(extract_reason);
316    let response = parsed_value.as_ref().map(|v| ErrorBody {
317        message: body_message.clone().unwrap_or_default(),
318        raw: Some(v.clone()),
319    });
320    match status.as_u16() {
321        401 => Error::Auth { response },
322        404 => Error::NotFound { response },
323        400 => Error::Validation {
324            message: extract_validation_message(parsed_value.as_ref()),
325            response,
326        },
327        429 => {
328            let retry_after = parse_retry_after(headers).as_secs();
329            Error::RateLimit {
330                retry_after: u32::try_from(retry_after).unwrap_or(u32::MAX),
331                limit_type: headers
332                    .get(RATE_LIMIT_TYPE_HDR)
333                    .and_then(|v| v.to_str().ok())
334                    .map(str::to_string),
335                response,
336            }
337        }
338        code => Error::Api {
339            status: code,
340            message: body_message
341                .unwrap_or_else(|| format!("API request failed with status {code}")),
342            response,
343        },
344    }
345}
346
347/// The server's own explanation from an error body, e.g. why a 403 refused the request.
348///
349/// The API sends either `{"detail": "..."}` or a structured body such as a tier gate's `{"error": "...", "code": "tier_required", ...}`, sometimes nested under `detail`.
350fn extract_reason(v: &Value) -> Option<String> {
351    let Value::Object(map) = v else { return None };
352    let envelope_reason = |map: &serde_json::Map<String, Value>| {
353        ["error", "message", "detail"]
354            .iter()
355            .find_map(|key| match map.get(*key) {
356                Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
357                _ => None,
358            })
359    };
360    envelope_reason(map).or_else(|| match map.get("detail") {
361        Some(Value::Object(nested)) => envelope_reason(nested),
362        _ => None,
363    })
364}
365
366/// Decode a JSON byte slice into a typed `T`, wrapping decode errors so they
367/// surface as [`Error::Decode`].
368pub(crate) fn decode_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
369    serde_json::from_slice(bytes).map_err(Error::Decode)
370}
371
372/// Decode a JSON byte slice, treating an empty body as a default-constructed
373/// value. Useful for endpoints that legally return `204 No Content`.
374pub(crate) fn decode_json_or_default<T>(bytes: &[u8]) -> Result<T>
375where
376    T: DeserializeOwned + Default,
377{
378    if bytes.is_empty() {
379        return Ok(T::default());
380    }
381    decode_json(bytes)
382}
383
384#[cfg(test)]
385mod tests {
386    use super::*;
387    use serde_json::json;
388
389    #[test]
390    fn extract_envelope_detail() {
391        let body = json!({"detail": "no soup for you"});
392        assert_eq!(
393            extract_validation_message(Some(&body)),
394            "invalid request parameters: no soup for you"
395        );
396    }
397
398    #[test]
399    fn extract_envelope_message() {
400        let body = json!({"message": "bad input"});
401        assert_eq!(
402            extract_validation_message(Some(&body)),
403            "invalid request parameters: bad input"
404        );
405    }
406
407    #[test]
408    fn extract_field_errors_sorted() {
409        let body = json!({
410            "zebra": ["last alphabetically"],
411            "apple": ["first alphabetically"],
412        });
413        // sorted keys => `apple` wins, deterministically.
414        assert_eq!(
415            extract_validation_message(Some(&body)),
416            "invalid request parameters: first alphabetically"
417        );
418    }
419
420    #[test]
421    fn extract_falls_back() {
422        let body = json!({});
423        assert_eq!(
424            extract_validation_message(Some(&body)),
425            "invalid request parameters"
426        );
427        assert_eq!(
428            extract_validation_message(None),
429            "invalid request parameters"
430        );
431    }
432
433    #[test]
434    fn extract_prefers_envelope_over_field() {
435        let body = json!({
436            "detail": "envelope wins",
437            "apple": ["field loses"],
438        });
439        assert_eq!(
440            extract_validation_message(Some(&body)),
441            "invalid request parameters: envelope wins"
442        );
443    }
444
445    #[test]
446    fn extract_string_field_error() {
447        let body = json!({"piid": "must be present"});
448        assert_eq!(
449            extract_validation_message(Some(&body)),
450            "invalid request parameters: must be present"
451        );
452    }
453
454    fn api_error(status: u16, body: &[u8]) -> (u16, String, Option<ErrorBody>, String) {
455        let err = decode_error(
456            StatusCode::from_u16(status).unwrap(),
457            &HeaderMap::new(),
458            body,
459        );
460        let display = err.to_string();
461        match err {
462            Error::Api {
463                status,
464                message,
465                response,
466            } => (status, message, response, display),
467            other => panic!("expected Error::Api, got {other:?}"),
468        }
469    }
470
471    #[test]
472    fn api_error_carries_the_server_reason() {
473        let body = json!({"detail": "These shape fields are not allowed for your account."});
474        let (status, message, response, display) = api_error(403, body.to_string().as_bytes());
475        assert_eq!(status, 403);
476        assert_eq!(
477            message,
478            "These shape fields are not allowed for your account."
479        );
480        assert_eq!(
481            display,
482            "tango: API error (status 403): These shape fields are not allowed for your account."
483        );
484        let response = response.unwrap();
485        assert_eq!(response.message, message);
486        assert_eq!(response.raw, Some(body));
487    }
488
489    #[test]
490    fn api_error_tier_gate_reason_flat_or_nested() {
491        let reason = "The 'psc' filter requires a Small plan or higher.";
492        for body in [
493            json!({"error": reason, "code": "tier_required"}),
494            json!({"detail": {"error": reason, "code": "tier_required"}}),
495        ] {
496            let (_, message, response, _) = api_error(403, body.to_string().as_bytes());
497            assert_eq!(message, reason, "body: {body}");
498            assert_eq!(response.unwrap().raw, Some(body));
499        }
500    }
501
502    #[test]
503    fn api_error_json_body_without_a_reason_keeps_the_body() {
504        let body = json!({"code": "forbidden"});
505        let (_, message, response, _) = api_error(403, body.to_string().as_bytes());
506        assert_eq!(message, "API request failed with status 403");
507        let response = response.unwrap();
508        assert_eq!(response.message, "");
509        assert_eq!(response.raw, Some(body));
510    }
511
512    #[test]
513    fn api_error_non_json_body_keeps_the_bare_message() {
514        let (status, message, response, _) = api_error(502, b"<html>Bad Gateway</html>");
515        assert_eq!(status, 502);
516        assert_eq!(message, "API request failed with status 502");
517        assert_eq!(response, None);
518    }
519
520    #[test]
521    fn backoff_doubles_and_caps() {
522        let base = Duration::from_millis(250);
523        assert_eq!(backoff_for(base, 0), Duration::from_millis(250));
524        assert_eq!(backoff_for(base, 1), Duration::from_millis(500));
525        assert_eq!(backoff_for(base, 2), Duration::from_secs(1));
526        assert_eq!(backoff_for(base, 3), Duration::from_secs(2));
527        assert_eq!(backoff_for(base, 4), Duration::from_secs(4));
528        assert_eq!(backoff_for(base, 5), Duration::from_secs(8));
529        assert_eq!(backoff_for(base, 6), MAX_BACKOFF);
530        assert_eq!(backoff_for(base, 50), MAX_BACKOFF);
531    }
532}
tango/transport.rs

tango/
transport.rs
