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