1use std::pin::Pin;
2use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
3use std::task::{Context, Poll};
4
5use bytes::Bytes;
6use futures_util::Stream;
7
8use super::RequestBuilder;
9use super::route::Route;
10use crate::core::Result;
11use crate::core::body::{Body, BodyKind, BoxedStream};
12use crate::core::deadline::Deadline;
13use crate::core::error::{Error, Kind};
14use crate::core::response::Response;
15use crate::core::session::execute::Attempt;
16use crate::core::transport::ResponseMode;
17use crate::util::is_idempotent;
18
19impl RequestBuilder {
20 pub async fn send(self) -> Result<Response> {
21 Box::pin(self.run()).await
22 }
23
24 async fn run(mut self) -> Result<Response> {
25 self.prepare()?;
26 let retry_policy = self.retry_policy.clone();
27 let session = self.session.clone();
28 let deadline = session.deadline(self.timeouts.as_ref());
29 let tag = self.tag.take();
30 let status_errors = self.status_errors;
31 let attempt = self.into_attempt(deadline);
32 let method = attempt.method.clone();
33 let url = attempt.url.clone();
34 let streamed = attempt.response == ResponseMode::Streamed;
35 let mut exhausted = false;
36 let response = session
37 .traced(
38 method.as_str(),
39 &url,
40 streamed,
41 run_attempts(
42 &session,
43 attempt,
44 &retry_policy,
45 deadline,
46 tag,
47 &mut exhausted,
48 ),
49 )
50 .await?;
51 if !status_errors {
52 return Ok(response);
53 }
54 let policy_wait = retry_policy.server_wait(&response);
55 response
56 .error_for_status_with_body(&deadline)
57 .await
58 .map_err(|mut error| {
59 error.set_retry_outcome(policy_wait, exhausted);
60 error
61 })
62 }
63
64 fn into_attempt(mut self, deadline: Deadline) -> Attempt {
65 let headers = if self.headers.is_empty() {
66 None
67 } else {
68 Some(std::mem::take(&mut self.headers))
69 };
70 Attempt {
71 method: std::mem::take(&mut self.method),
72 url: std::mem::take(&mut self.url),
73 preset: self.preset,
74 body: std::mem::take(&mut self.body),
75 headers,
76 deadline,
77 response: ResponseMode::new(
78 self.stream_response,
79 self.status_errors
80 .then(|| self.session.error_budget(&deadline)),
81 ),
82 proxy: self.proxy.take(),
83 header_order: self.header_order.take(),
84 redirect: self.redirect.take(),
85 digest: self.digest_auth.take(),
86 initiator: self.initiator.take(),
87 trusted_origin: self.trusted_origin.take(),
88 }
89 }
90
91 pub(crate) fn prepare(&mut self) -> Result<()> {
92 if let Some(err) = self.builder_error.take() {
93 return Err(err);
94 }
95 self.infer_from_content_type();
96
97 if !self.query_params.is_empty() {
98 let mut url = url::Url::parse(&self.url).map_err(crate::core::Error::from_url_parse)?;
99 {
100 let mut pairs = url.query_pairs_mut();
101 for (k, v) in &self.query_params {
102 pairs.append_pair(k, v);
103 }
104 }
105 self.url = url.to_string();
106 }
107
108 if let Some(encoding) = self.compress {
109 match &self.body.0 {
110 BodyKind::Bytes(b) => {
111 let compressed = encoding.encode(b)?;
112 self.headers
113 .set("content-encoding", encoding.header_value())?;
114 self.body = Body::from(compressed);
115 }
116 BodyKind::Empty => {}
117 BodyKind::Stream { .. } => {
118 return Err(Error::new(Kind::Body).with_message(
119 "request-body compression is not supported for streaming bodies; \
120 buffer the body into bytes before calling `.compress(..)`",
121 ));
122 }
123 }
124 }
125 Ok(())
126 }
127}
128
129async fn run_attempts(
130 session: &crate::core::session::Session,
131 mut attempt: Attempt,
132 retry_policy: &crate::core::retry::RetryPolicy,
133 deadline: Deadline,
134 tag: Option<String>,
135 exhausted: &mut bool,
136) -> (Result<Response>, u32) {
137 crate::trace::note_tag(tag);
138 let mut route = Route::new(session, &attempt, retry_policy);
139 if retry_policy.is_none() {
140 return (counted(route.send(attempt, None).await, 1), 1);
141 }
142 let retryable_method =
143 retry_policy.allow_non_idempotent || is_idempotent(attempt.method.as_str());
144 let replay = attempt.body.replay();
145 let mut rewind = Rewind::wrap(&mut attempt.body);
146 let base_headers = attempt.headers.clone();
147
148 let mut n: u32 = 0;
149 loop {
150 if let Err(err) = deadline.check() {
151 return (Err(err), n);
152 }
153 let attempt_body = std::mem::take(&mut attempt.body);
154 let this = attempt.again(attempt_body, base_headers.clone());
155 let result = route.send(this, retry_policy.proxy_for_retry(n)).await;
156
157 let body_retryable = || replay.is_some() || rewind.as_mut().is_some_and(Rewind::reclaim);
158 let sleep = match plan_retry(&result, retry_policy, n, retryable_method, body_retryable) {
159 RetryPlan::Backoff(sleep) if sleep < deadline.remaining() => sleep,
160 plan => {
161 *exhausted = matches!(
162 plan,
163 RetryPlan::Exhausted | RetryPlan::Declined | RetryPlan::Backoff(_)
164 );
165 return (counted(result, n + 1), n + 1);
166 }
167 };
168 drop(result);
169
170 let slept = session
171 .unless_shut_down(async {
172 deadline.sleep(sleep).await;
173 Ok(())
174 })
175 .await;
176 if let Err(err) = slept {
177 return (counted(Err(err), n + 1), n + 1);
178 }
179 n += 1;
180 attempt.body = replay
181 .as_ref()
182 .and_then(Body::replay)
183 .or_else(|| rewind.as_mut().and_then(Rewind::again))
184 .unwrap_or_default();
185 }
186}
187
188fn counted(result: Result<Response>, attempts: u32) -> Result<Response> {
189 result
190 .map(|mut response| {
191 response.set_attempts(attempts);
192 response
193 })
194 .map_err(|mut error| {
195 error.set_attempts(attempts);
196 error
197 })
198}
199
200enum RetryPlan {
201 Stop,
202 Exhausted,
203 Declined,
204 Backoff(std::time::Duration),
205}
206
207fn plan_retry(
208 result: &Result<Response>,
209 retry_policy: &crate::core::retry::RetryPolicy,
210 attempt: u32,
211 retryable_method: bool,
212 body_retryable: impl FnOnce() -> bool,
213) -> RetryPlan {
214 if retry_policy.is_none() {
215 return RetryPlan::Stop;
216 }
217 if attempt >= retry_policy.max_retries {
218 return match result {
219 Ok(response) if retry_policy.matches_response(response) => RetryPlan::Exhausted,
220 _ => RetryPlan::Stop,
221 };
222 }
223 if !retry_wanted(result, retry_policy)
224 || !(retryable_method || retry_policy.retries_unsent(result))
225 || !body_retryable()
226 {
227 return RetryPlan::Stop;
228 }
229 match result
230 .as_ref()
231 .ok()
232 .and_then(|r| retry_policy.server_wait(r))
233 {
234 Some(wait) if wait > retry_policy.max_retry_after => RetryPlan::Declined,
235 Some(wait) => RetryPlan::Backoff(wait),
236 None => RetryPlan::Backoff(retry_policy.backoff(attempt)),
237 }
238}
239
240fn retry_wanted(result: &Result<Response>, retry_policy: &crate::core::retry::RetryPolicy) -> bool {
241 match result {
242 Ok(resp) => retry_policy.matches_response(resp),
243 Err(err) if err.is_timeout() => retry_policy.matches_timeout(),
244 Err(err) if err.is_retryable() => retry_policy.matches_connection_error(),
245 Err(_) => false,
246 }
247}
248
249struct Slot {
250 stream: Option<BoxedStream>,
251 polled: bool,
252}
253
254type SharedSlot = Arc<Mutex<Slot>>;
255
256fn lock(slot: &SharedSlot) -> MutexGuard<'_, Slot> {
257 slot.lock().unwrap_or_else(PoisonError::into_inner)
258}
259
260struct Rewind {
261 slot: SharedSlot,
262 length_hint: Option<u64>,
263 held: Option<BoxedStream>,
264}
265
266impl Rewind {
267 fn wrap(body: &mut Body) -> Option<Self> {
268 if !matches!(body.0, BodyKind::Stream { .. }) {
269 return None;
270 }
271 let BodyKind::Stream {
272 stream,
273 length_hint,
274 } = std::mem::take(&mut body.0)
275 else {
276 return None;
277 };
278 let rewind = Self {
279 slot: Arc::new(Mutex::new(Slot {
280 stream: Some(stream),
281 polled: false,
282 })),
283 length_hint,
284 held: None,
285 };
286 *body = rewind.body();
287 Some(rewind)
288 }
289
290 fn body(&self) -> Body {
291 Body(BodyKind::Stream {
292 stream: Box::pin(Reclaim(Arc::clone(&self.slot))),
293 length_hint: self.length_hint,
294 })
295 }
296
297 fn reclaim(&mut self) -> bool {
298 let mut slot = lock(&self.slot);
299 if slot.polled {
300 return false;
301 }
302 self.held = slot.stream.take();
303 self.held.is_some()
304 }
305
306 fn again(&mut self) -> Option<Body> {
307 let stream = self.held.take()?;
308 self.slot = Arc::new(Mutex::new(Slot {
309 stream: Some(stream),
310 polled: false,
311 }));
312 Some(self.body())
313 }
314}
315
316struct Reclaim(SharedSlot);
317
318impl Stream for Reclaim {
319 type Item = std::io::Result<Bytes>;
320
321 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
322 let mut slot = lock(&self.0);
323 slot.polled = true;
324 match slot.stream.as_mut() {
325 Some(stream) => stream.as_mut().poll_next(cx),
326 None => Poll::Ready(None),
327 }
328 }
329}
330
331impl std::future::IntoFuture for RequestBuilder {
332 type Output = Result<Response>;
333 type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send>>;
334
335 fn into_future(self) -> Self::IntoFuture {
336 Box::pin(async move { self.send().await })
337 }
338}
339
340#[cfg(test)]
341#[path = "send_tests.rs"]
342mod tests;