polyoxide_core/send.rs
1//! [`HttpClient::send`]: core's one send loop.
2
3use serde::de::DeserializeOwned;
4use url::Url;
5
6use crate::client::HttpClient;
7use crate::error::{ApiError, ErrorResponse};
8use crate::hooks::{
9 Authenticator, Cost, DynAuthenticator, Outcome, RequestMeta, RequestParts, ResponseMeta,
10 RetryPolicy, Throttle,
11};
12use crate::{response::Response, transport::HttpRequest};
13
14impl HttpClient {
15 /// Send `parts`, retrying as the client's policy decides.
16 ///
17 /// Each attempt takes a concurrency permit, waits while the throttle
18 /// charges it ([`Throttle::acquire`], with `costs`), is signed by `auth`
19 /// when one is given, and is sent, within `parts.timeout` when it is set.
20 /// The throttle then sees the response
21 /// ([`Throttle::observe`]), the policy decides what follows
22 /// ([`RetryPolicy::decide`]), and a hold the decision carries is applied
23 /// ([`Throttle::hold`]) whether or not the request is retried.
24 ///
25 /// A retry releases the permit and sleeps the longer of the policy's wait
26 /// and the loop's own floor, [`RetryConfig::retry_delay`] for the attempt.
27 /// The floor belongs to the loop, so no policy can retry sooner. The loop
28 /// never retries past [`RetryConfig::max_retries`], whatever the policy
29 /// says.
30 ///
31 /// Each retry logs a `WARN` under the `polyoxide_core` target,
32 /// `Retriable status <code> on <path>, retry <n> after <ms>ms`, where
33 /// `<ms>` is the longer of the retry's sleep and the hold its decision
34 /// set, since the retry waits out both. A hold that is not a retry logs
35 /// one too: `Status <code> on <path>, not retried: every request held
36 /// <ms>ms`.
37 ///
38 /// Returns the response the policy is done with, whatever its status; one
39 /// it fails, or would retry with no retry left, is an error.
40 ///
41 /// # Errors
42 ///
43 /// - [`ApiError::Response`] when the policy fails the last response.
44 /// - [`ApiError::Url`] when `parts.path` does not join the base URL.
45 /// - [`ApiError::Refused`] when the throttle refuses a cost, or the
46 /// signing error when `auth` fails. Nothing is sent.
47 /// - [`ApiError::Transport`] when no response arrives. That attempt is
48 /// neither observed nor retried.
49 ///
50 /// [`RetryConfig::retry_delay`]: crate::RetryConfig::retry_delay
51 /// [`RetryConfig::max_retries`]: crate::RetryConfig::max_retries
52 pub async fn send(
53 &self,
54 parts: RequestParts,
55 costs: &[Cost],
56 auth: Option<&DynAuthenticator<'_>>,
57 ) -> Result<Response, ApiError> {
58 let url = self.base_url.join(&parts.path)?;
59 let mut attempt = 0u32;
60
61 loop {
62 let permit = self.acquire_concurrency().await;
63 let request = RequestMeta {
64 method: &parts.method,
65 path: &parts.path,
66 query: &parts.query,
67 costs,
68 };
69 let charge = self.throttle.acquire(&request).await?;
70
71 // Signed afresh on every attempt, so a timestamp in it is current.
72 let mut signed = parts.clone();
73 if let Some(auth) = auth {
74 auth.sign(&mut signed, attempt).await?;
75 }
76 let response = self.transport.send(self.http_request(&url, signed)).await?;
77
78 let status = response.status();
79 let info = self.retry_config.attempt_info(attempt);
80 let meta = ResponseMeta {
81 status,
82 headers: response.headers(),
83 };
84 self.throttle.observe(&charge, &meta, &info);
85 let decision = self.policy.decide(&meta, &info, &self.retry_config);
86
87 // Before the retry decision, and whatever it is: a hold is a fact
88 // about the host, and a request with no retry left still has to
89 // publish it to every request sharing this throttle.
90 if let Some(hold) = decision.hold {
91 self.throttle.hold(hold);
92 }
93
94 match decision.outcome {
95 Outcome::Retry(wait) if info.retries_left > 0 => {
96 let floor = self.retry_config.retry_delay(attempt, meta.retry_after());
97 let sleep = floor.max(wait);
98 attempt += 1;
99 // The retry also waits out the hold this decision set, so
100 // the logged wait is the longer of the two.
101 tracing::warn!(
102 "Retriable status {} on {}, retry {} after {}ms",
103 status,
104 parts.path,
105 attempt,
106 decision
107 .hold
108 .map_or(sleep, |hold| hold.max(sleep))
109 .as_millis()
110 );
111 drop(permit);
112 self.runtime.sleep(sleep).await;
113 }
114 Outcome::Done | Outcome::Fail | Outcome::Retry(_) => {
115 // A hold outlives this request, so it is logged even when
116 // nothing is retried (DRIFT R10).
117 if let Some(hold) = decision.hold {
118 tracing::warn!(
119 "Status {} on {}, not retried: every request held {}ms",
120 status,
121 parts.path,
122 hold.as_millis()
123 );
124 }
125 // `Done` hands the response back whatever its status. A
126 // `Fail`, or a retry with none left, is the error AD-8
127 // names: the status, headers, body and `Retry-After`.
128 return match decision.outcome {
129 Outcome::Done => Ok(response),
130 Outcome::Fail | Outcome::Retry(_) => {
131 // Read outside the permit, as a caller reads a
132 // 2xx body after the loop drops it, so a slow
133 // error body holds no slot.
134 drop(permit);
135 Err(ErrorResponse::read(response).await.into())
136 }
137 };
138 }
139 }
140 }
141 }
142
143 /// One attempt's request, built from its signed parts: the URL with the
144 /// query form-urlencoded onto it, as reqwest's `.query()` did, and a
145 /// timeout that is always set, the request's own else the client's.
146 fn http_request(&self, url: &Url, parts: RequestParts) -> HttpRequest {
147 let mut url = url.clone();
148 if !parts.query.is_empty() {
149 url.query_pairs_mut().extend_pairs(&parts.query);
150 }
151 HttpRequest {
152 method: parts.method,
153 url,
154 headers: parts.headers,
155 body: parts.body,
156 timeout: parts.timeout.unwrap_or(self.timeout),
157 }
158 }
159}
160
161/// Decode a response body as JSON, logging a failure once.
162///
163/// A failure logs one ERROR line under `polyoxide_core`,
164/// `Failed to decode {path}: {err}: {body}`, with the body truncated for the
165/// log, and returns the error for the caller to map into its own type.
166pub fn decode_json<T: DeserializeOwned>(path: &str, text: &str) -> Result<T, serde_json::Error> {
167 serde_json::from_str(text).map_err(|err| {
168 tracing::error!(
169 "Failed to decode {path}: {err}: {}",
170 crate::truncate_for_log(text)
171 );
172 err
173 })
174}