rp_supabase_http/
lib.rs

1#![cfg_attr(doc, doc = include_str!("../README.md"))]
2
3use core::time::Duration;
4use std::time::SystemTime;
5
6/// Start a client builder with the selected TLS backend and same-origin redirects.
7/// rustls takes precedence when both TLS features are enabled.
8pub fn client_builder() -> reqwest::ClientBuilder {
9    let builder =
10        reqwest::Client::builder().redirect(reqwest::redirect::Policy::custom(|attempt| {
11            if !attempt
12                .previous()
13                .last()
14                .is_some_and(|url| url.origin() == attempt.url().origin())
15            {
16                attempt.stop()
17            } else if attempt.previous().len() > 10 {
18                attempt.error("too many redirects")
19            } else {
20                attempt.follow()
21            }
22        }));
23    #[cfg(feature = "rustls")]
24    let builder = builder.tls_backend_rustls();
25    #[cfg(all(feature = "native-tls", not(feature = "rustls")))]
26    let builder = builder.tls_backend_native();
27    builder
28}
29
30/// A bounded retry policy for explicit HTTP 429 responses.
31///
32/// Transport failures are never retried. The default replays only GET, HEAD and
33/// OPTIONS requests, and only when their bodies can be cloned by reqwest.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct RetryPolicy {
36    max_retries: u8,
37    initial_delay: Duration,
38    max_delay: Duration,
39    non_idempotent: bool,
40}
41
42impl Default for RetryPolicy {
43    fn default() -> Self {
44        Self::new(2, Duration::from_millis(200), Duration::from_secs(30))
45    }
46}
47
48impl RetryPolicy {
49    /// Set the number of additional attempts and the backoff bounds.
50    /// A Retry-After value above `max_delay` returns the 429 without waiting.
51    #[must_use]
52    pub const fn new(max_retries: u8, initial_delay: Duration, max_delay: Duration) -> Self {
53        Self {
54            max_retries,
55            initial_delay,
56            max_delay,
57            non_idempotent: false,
58        }
59    }
60
61    /// Send every request exactly once.
62    #[must_use]
63    pub const fn none() -> Self {
64        Self::new(0, Duration::ZERO, Duration::ZERO)
65    }
66
67    /// Permit replay of mutation requests with cloneable bodies after a 429.
68    ///
69    /// Enable only when the endpoint rejects rate-limited requests before any
70    /// side effect. This is not a general idempotency guarantee for POST or PATCH.
71    #[must_use]
72    pub const fn retry_non_idempotent(mut self, enabled: bool) -> Self {
73        self.non_idempotent = enabled;
74        self
75    }
76
77    const fn permits(self, method: &reqwest::Method) -> bool {
78        self.max_retries != 0
79            && (self.non_idempotent
80                || matches!(
81                    *method,
82                    reqwest::Method::GET | reqwest::Method::HEAD | reqwest::Method::OPTIONS
83                ))
84    }
85
86    fn delay(self, response: &reqwest::Response, attempt: u8) -> Option<Duration> {
87        let backoff = self
88            .initial_delay
89            .saturating_mul(2_u32.saturating_pow(u32::from(attempt)))
90            .min(self.max_delay);
91        let delay = match response
92            .headers()
93            .get(reqwest::header::RETRY_AFTER)
94            .and_then(|value| value.to_str().ok())
95            .map(str::trim)
96        {
97            Some(value) if !value.is_empty() && value.bytes().all(|byte| byte.is_ascii_digit()) => {
98                Duration::from_secs(value.parse().ok()?)
99            }
100            Some(value) => httpdate::parse_http_date(value)
101                .ok()
102                .map_or(backoff, |date| {
103                    date.duration_since(SystemTime::now())
104                        .unwrap_or(Duration::ZERO)
105                }),
106            None => backoff,
107        };
108        (delay <= self.max_delay).then_some(delay)
109    }
110}
111
112/// Execute a request with the supplied rate-limit retry policy.
113///
114/// The final response is unchanged, including its headers and body. Streaming
115/// bodies that reqwest cannot clone are sent once. Dropping this future cancels
116/// an in-flight request or a pending backoff.
117///
118/// # Errors
119/// Returns reqwest's transport error immediately, without replaying the request.
120pub async fn execute(
121    client: &reqwest::Client,
122    mut request: reqwest::Request,
123    policy: RetryPolicy,
124) -> Result<reqwest::Response, reqwest::Error> {
125    if !policy.permits(request.method()) {
126        return client.execute(request).await;
127    }
128    let Some(template) = request.try_clone() else {
129        return client.execute(request).await;
130    };
131    let mut attempt = 0;
132    loop {
133        let response = client.execute(request).await?;
134        if response.status() != reqwest::StatusCode::TOO_MANY_REQUESTS
135            || attempt >= policy.max_retries
136        {
137            return Ok(response);
138        }
139        let Some(delay) = policy.delay(&response, attempt) else {
140            return Ok(response);
141        };
142        let Some(next) = template.try_clone() else {
143            return Ok(response);
144        };
145        drop(response);
146        tokio::time::sleep(delay).await;
147        attempt = attempt.saturating_add(1);
148        request = next;
149    }
150}
rp_supabase_http/lib.rs

rp_supabase_http/
lib.rs
