vantage_api_pool/resilient/
mod.rs

1//! `ResilientClient` — an async-native HTTP transport.
2//!
3//! One client per remote API. Concurrency is bounded by a semaphore; retry,
4//! auth refresh, a rate limit and a circuit breaker are inline middleware.
5//! Cancellation is structural: drop the future and the in-flight request goes
6//! with it.
7//!
8//! The pieces live in one file each: `builder` assembles a client, `attempt`
9//! runs one call's attempt loop, `breaker` is the circuit breaker, `rate` the
10//! token bucket, `policy` the retry knobs, and `auth`, `error` and `observer`
11//! the types those exchange.
12
13mod attempt;
14mod auth;
15mod breaker;
16mod builder;
17mod error;
18mod observer;
19mod policy;
20mod rate;
21
22pub use auth::{AuthFuture, AuthRefresher};
23pub use breaker::BreakerState;
24pub use builder::ResilientClientBuilder;
25pub use error::{ClientError, ErrorKind};
26pub use observer::{TransportEvent, TransportObserver};
27pub use policy::{BreakerMode, CallPolicy, RetryMode, RetryPolicy};
28
29use std::sync::atomic::{AtomicUsize, Ordering};
30use std::sync::Arc;
31
32use tokio::sync::Semaphore;
33
34use attempt::AttemptLoop;
35use auth::AuthState;
36use breaker::CircuitBreaker;
37
38/// A cheap-to-clone resilient HTTP client. Build via [`ResilientClient::builder`].
39#[derive(Clone)]
40pub struct ResilientClient {
41    http: reqwest::Client,
42    semaphore: Arc<Semaphore>,
43    policy: RetryPolicy,
44    breaker: Option<Arc<CircuitBreaker>>,
45    auth: Option<Arc<AuthState>>,
46    in_flight: Arc<AtomicUsize>,
47    peak_in_flight: Arc<AtomicUsize>,
48    observer: Option<(Arc<str>, Arc<dyn TransportObserver>)>,
49    rate: Option<Arc<rate::TokenBucket>>,
50}
51
52impl ResilientClient {
53    pub fn builder() -> ResilientClientBuilder {
54        ResilientClientBuilder::default()
55    }
56
57    /// Execute a request with the client's default policies (bounded retry,
58    /// fail fast on an open breaker). `build` is called once per attempt.
59    pub async fn execute<F>(&self, build: F) -> Result<reqwest::Response, ClientError>
60    where
61        F: Fn(&reqwest::Client) -> reqwest::RequestBuilder,
62    {
63        let policy = CallPolicy::bounded(self.policy.clone());
64        self.execute_with(&policy, build).await
65    }
66
67    /// Execute a request under `policy`. Returns the first `2xx` response,
68    /// or the last attempt's failure. Dropping the returned future cancels
69    /// the in-flight request and any pending back-off.
70    pub async fn execute_with<F>(
71        &self,
72        policy: &CallPolicy,
73        build: F,
74    ) -> Result<reqwest::Response, ClientError>
75    where
76        F: Fn(&reqwest::Client) -> reqwest::RequestBuilder,
77    {
78        AttemptLoop::run(self, policy, &build).await
79    }
80
81    /// Requests in flight right now — sent, not yet answered. Waiting for a
82    /// breaker cooldown, a rate-limit token, a permit or a retry back-off
83    /// does not count.
84    pub fn in_flight(&self) -> usize {
85        self.in_flight.load(Ordering::SeqCst)
86    }
87
88    /// Highest number of simultaneously in-flight requests observed.
89    pub fn peak_in_flight(&self) -> usize {
90        self.peak_in_flight.load(Ordering::SeqCst)
91    }
92
93    /// What the circuit breaker is doing right now, for consumers that poll
94    /// rather than follow `TransportEvent`s. `None` when this client has no
95    /// breaker configured.
96    pub fn breaker_state(&self) -> Option<BreakerState> {
97        self.breaker.as_ref().map(|b| b.state())
98    }
99
100    /// The datasource key this client reports under, when it has an observer.
101    pub fn key(&self) -> Option<&str> {
102        self.observer.as_ref().map(|(k, _)| &**k)
103    }
104
105    /// Emit an event on behalf of the caller (`RowsPulled`, `WritePushed`).
106    pub fn report(&self, event: TransportEvent) {
107        if let Some((key, obs)) = &self.observer {
108            obs.on_event(key, event);
109        }
110    }
111
112    /// Count one request as in flight until the returned guard is dropped —
113    /// including when the caller's future is cancelled mid-send, which is why
114    /// this is a guard and not a pair of `fetch_add` / `fetch_sub` calls.
115    fn enter_flight(&self) -> InFlightGuard {
116        let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
117        self.peak_in_flight.fetch_max(now, Ordering::SeqCst);
118        InFlightGuard(Arc::clone(&self.in_flight))
119    }
120}
121
122impl std::fmt::Debug for ResilientClient {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        f.debug_struct("ResilientClient")
125            .field("key", &self.key())
126            .field("in_flight", &self.in_flight())
127            .finish_non_exhaustive()
128    }
129}
130
131struct InFlightGuard(Arc<AtomicUsize>);
132
133impl Drop for InFlightGuard {
134    fn drop(&mut self) {
135        self.0.fetch_sub(1, Ordering::SeqCst);
136    }
137}
vantage_api_pool/resilient/mod.rs

vantage_api_pool/resilient/
mod.rs
